Repository: vitessio/vitess
Pull request #10715 (2022-07-17 11:02:52)

BugFix: Keep predicates in join when pushing new ones
<!--
  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 fixes the bug found in https://github.com/vitessio/vitess/issues/10713. 
The plan for the query found there with Gen4 using the vschema provided was - 
```json
{
  "QueryType": "SELECT",
  "Original": "select user.id, team.id from user join team_member on user.id = team_member.user join team on team_member.team = team.id where user.email = 'alice@gmail.com'",
  "Instructions": {
    "OperatorType": "Join",
    "Variant": "Join",
    "JoinColumnIndexes": "L:0,R:0",
    "JoinVars": {
      "user_id": 0
    },
    "TableName": "`user`_team, team_member",
    "Inputs": [
      {
        "OperatorType": "Route",
        "Variant": "EqualUnique",
        "Keyspace": {
          "Name": "customer",
          "Sharded": true
        },
        "FieldQuery": "select `user`.id from `user` where 1 != 1",
        "Query": "select `user`.id from `user` where `user`.email = 'alice@gmail.com'",
        "Table": "`user`",
        "Values": [
          "VARCHAR(\"alice@gmail.com\")"
        ],
        "Vindex": "email_user_map"
      },
      {
        "OperatorType": "Route",
        "Variant": "Scatter",
        "Keyspace": {
          "Name": "customer",
          "Sharded": true
        },
        "FieldQuery": "select team.id from team_member, team where 1 != 1",
        "Query": "select team.id from team_member, team where team_member.`user` = :user_id",
        "Table": "team, team_member"
      }
    ]
  }
}
```

In the plan, it is clearly visible that the predicate `team_member.team = team.id` is not applied anywhere. This results in extra rows being returned due to insufficient filtering as reported by the user.

After the changes in the PR, the plan looks like 
```json
{
  "QueryType": "SELECT",
  "Original": "select user.id, team.id from user join team_member on user.id = team_member.user join team on team_member.team = team.id where user.email = 'alice@gmail.com'",
  "Instructions": {
    "OperatorType": "Join",
    "Variant": "Join",
    "JoinColumnIndexes": "L:0,R:0",
    "JoinVars": {
      "user_id": 0
    },
    "TableName": "`user`_team, team_member",
    "Inputs": [
      {
        "OperatorType": "Route",
        "Variant": "EqualUnique",
        "Keyspace": {
          "Name": "customer",
          "Sharded": true
        },
        "FieldQuery": "select `user`.id from `user` where 1 != 1",
        "Query": "select `user`.id from `user` where `user`.email = 'alice@gmail.com'",
        "Table": "`user`",
        "Values": [
          "VARCHAR(\"alice@gmail.com\")"
        ],
        "Vindex": "email_user_map"
      },
      {
        "OperatorType": "Route",
        "Variant": "Scatter",
        "Keyspace": {
          "Name": "customer",
          "Sharded": true
        },
        "FieldQuery": "select team.id from team_member, team where 1 != 1",
        "Query": "select team.id from team_member, team where team_member.`user` = :user_id and team_member.team = team.id",
        "Table": "team, team_member"
      }
    ]
  }
}
```
Now, the predicate is found in the last route.

The earliest release affected by the bug is 13.0 and as such, the PR should be backported to both release-14.0 and release-13.0

## Related Issue(s)

<!-- List related issues and pull requests. If this PR fixes an issue, please add it using Fixes #????  -->
- https://github.com/vitessio/vitess/issues/10713

## Checklist

-   [x] "Backport me!" label has been added if this change should be backported - Both to release-13.0 and release-14.0
-   [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. -->


vitess-bot[bot]: ### Review Checklist
            
Hello reviewers! :wave: Please follow this checklist when reviewing this Pull Request.

#### General
- [ ] Ensure that the Pull Request has a descriptive title.
- [ ] If this is a change that users need to know about, please apply the `release notes (needs details)` label so that merging is blocked unless the summary release notes document is included.
- [ ] If a new flag is being introduced, review whether it is really needed. The flag names should be clear and intuitive (as far as possible), and the flag's help should be descriptive. Additionally, flag names should use dashes (`-`) as word separators rather than underscores (`_`).
- [ ] If a workflow is added or modified, each items in `Jobs` should be named in order to mark it as `required`. If the workflow should be required, the GitHub Admin should be notified.

#### Bug fixes
- [ ] There should be at least one unit or end-to-end test.
- [ ] The Pull Request description should either include a link to an issue that describes the bug OR an actual description of the bug and how to reproduce, along with a description of the fix.

#### Non-trivial changes
- [ ] There should be some code comments as to why things are implemented the way they are.

#### New/Existing features
- [ ] Should be documented, either by modifying the existing documentation or creating new documentation.
- [ ] New features should have a link to a feature request issue or an RFC that documents the use cases, corner cases and test cases.

#### Backward compatibility            
- [ ] Protobuf changes should be wire-compatible.
- [ ] Changes to `_vt` tables and RPCs need to be backward compatible.
- [ ] `vtctl` command output order should be stable and `awk`-able.

harshit-gangal: The changes looks good.
Can you add an e2e test as well?

I’m not sure about v13 backport as Gen4 was experimental on that.

Patch:
diff --git a/doc/releasenotes/14_0_0_release_notes.md b/doc/releasenotes/14_0_0_release_notes.md
index ad935e1ce8..50d02232ae 100644
--- a/doc/releasenotes/14_0_0_release_notes.md
+++ b/doc/releasenotes/14_0_0_release_notes.md
@@ -15,6 +15,10 @@
 - [Advisory locking optimizations](#advisory-locking-optimizations)
 - [Pre-Legacy Resharding is now deprecated](#pre-legacy-resharding-is-now-deprecated)
 
+## Known Issues
+
+- [VTOrc doesn't discover the tablets](https://github.com/vitessio/vitess/issues/10650) of a keyspace if the durability policy doesn't exist in the topo server when it comes up. This can be resolved by restarting VTOrc.
+
 ## Major Changes
 
 ### Gen4 is now the default planner
diff --git a/doc/releasenotes/14_0_0_summary.md b/doc/releasenotes/14_0_0_summary.md
index 1e65bd97c1..e02c9d9a28 100644
--- a/doc/releasenotes/14_0_0_summary.md
+++ b/doc/releasenotes/14_0_0_summary.md
@@ -14,6 +14,10 @@
 - [Advisory locking optimizations](#advisory-locking-optimizations)
 - [Pre-Legacy Resharding is now deprecated](#pre-legacy-resharding-is-now-deprecated)
 
+## Known Issues
+
+- [VTOrc doesn't discover the tablets](https://github.com/vitessio/vitess/issues/10650) of a keyspace if the durability policy doesn't exist in the topo server when it comes up. This can be resolved by restarting VTOrc.
+
 ## Major Changes
 
 ### Gen4 is now the default planner
diff --git a/go/mysql/collations/env.go b/go/mysql/collations/env.go
index 2dbf7e5e4c..a0873fa067 100644
--- a/go/mysql/collations/env.go
+++ b/go/mysql/collations/env.go
@@ -125,13 +125,17 @@ func fetchCacheEnvironment(version collver) *Environment {
 // The version string must be in the format that is sent by the server as the version packet
 // when opening a new MySQL connection
 func NewEnvironment(serverVersion string) *Environment {
-	var version collver = collverMySQL56
+	// 5.7 is the oldest version we support today, so use that as
+	// the default.
+	// NOTE: this should be changed when we EOL MySQL 5.7 support
+	var version collver = collverMySQL57
+	serverVersion = strings.TrimSpace(strings.ToLower(serverVersion))
 	switch {
 	case strings.HasSuffix(serverVersion, "-ripple"):
 		// the ripple binlog server can mask the actual version of mysqld;
 		// assume we have the highest
 		version = collverMySQL80
-	case strings.Contains(serverVersion, "MariaDB"):
+	case strings.Contains(serverVersion, "mariadb"):
 		switch {
 		case strings.Contains(serverVersion, "10.0."):
 			version = collverMariaDB100
diff --git a/go/mysql/collations/local.go b/go/mysql/collations/local.go
index a656a3dad2..7c3b2ab0b1 100644
--- a/go/mysql/collations/local.go
+++ b/go/mysql/collations/local.go
@@ -36,7 +36,10 @@ func Local() *Environment {
 			panic("collations.Local() called too early")
 		}
 		if *servenv.MySQLServerVersion == "" {
-			defaultEnv = fetchCacheEnvironment(collverMySQL80)
+			// 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)
 		}
diff --git a/go/vt/schemamanager/tablet_executor.go b/go/vt/schemamanager/tablet_executor.go
index 7ab3f5d3eb..7cf3945286 100644
--- a/go/vt/schemamanager/tablet_executor.go
+++ b/go/vt/schemamanager/tablet_executor.go
@@ -276,10 +276,10 @@ func (exec *TabletExecutor) preflightSchemaChanges(ctx context.Context, sqls []s
 
 // executeSQL executes a single SQL statement either as online DDL or synchronously on all tablets.
 // In online DDL case, the query may be exploded into multiple queries during
-func (exec *TabletExecutor) executeSQL(ctx context.Context, sql string, providedUUID string, execResult *ExecuteResult) error {
+func (exec *TabletExecutor) executeSQL(ctx context.Context, sql string, providedUUID string, execResult *ExecuteResult) (executedAsynchronously bool, err error) {
 	stmt, err := sqlparser.Parse(sql)
 	if err != nil {
-		return err
+		return false, err
 	}
 	switch stmt := stmt.(type) {
 	case sqlparser.DDLStatement:
@@ -287,7 +287,7 @@ func (exec *TabletExecutor) executeSQL(ctx context.Context, sql string, provided
 			onlineDDLs, err := schema.NewOnlineDDLs(exec.keyspace, sql, stmt, exec.ddlStrategySetting, exec.migrationContext, providedUUID)
 			if err != nil {
 				execResult.ExecutorErr = err.Error()
-				return err
+				return false, err
 			}
 			for _, onlineDDL := range onlineDDLs {
 				exec.executeOnAllTablets(ctx, execResult, onlineDDL.SQL, true)
@@ -296,25 +296,25 @@ func (exec *TabletExecutor) executeSQL(ctx context.Context, sql string, provided
 					exec.logger.Printf("%s\n", onlineDDL.UUID)
 				}
 			}
-			return nil
+			return true, nil
 		}
 	case *sqlparser.RevertMigration:
 		strategySetting := schema.NewDDLStrategySetting(schema.DDLStrategyOnline, exec.ddlStrategySetting.Options)
 		onlineDDL, err := schema.NewOnlineDDL(exec.keyspace, "", sqlparser.String(stmt), strategySetting, exec.migrationContext, providedUUID)
 		if err != nil {
 			execResult.ExecutorErr = err.Error()
-			return err
+			return false, err
 		}
 		exec.executeOnAllTablets(ctx, execResult, onlineDDL.SQL, true)
 		execResult.UUIDs = append(execResult.UUIDs, onlineDDL.UUID)
 		exec.logger.Printf("%s\n", onlineDDL.UUID)
-		return nil
+		return true, nil
 	case *sqlparser.AlterMigration:
 		exec.executeOnAllTablets(ctx, execResult, sql, true)
-		return nil
+		return true, nil
 	}
 	exec.executeOnAllTablets(ctx, execResult, sql, false)
-	return nil
+	return false, nil
 }
 
 // Execute applies schema changes
@@ -357,19 +357,61 @@ func (exec *TabletExecutor) Execute(ctx context.Context, sqls []string) *Execute
 	}
 	providedUUID := ""
 
+	syncOperationExecuted := false
 	for index, sql := range sqls {
 		execResult.CurSQLIndex = index
 		if exec.hasProvidedUUIDs() {
 			providedUUID = exec.uuids[index]
 		}
-		if err := exec.executeSQL(ctx, sql, providedUUID, &execResult); err != nil {
+		executedAsynchronously, err := exec.executeSQL(ctx, sql, providedUUID, &execResult)
+		if err != nil {
 			execResult.ExecutorErr = err.Error()
 			return &execResult
 		}
+		if !executedAsynchronously {
+			syncOperationExecuted = true
+		}
 		if len(execResult.FailedShards) > 0 {
 			break
 		}
 	}
+
+	if syncOperationExecuted {
+		// same shards will appear multiple times in execResult.SuccessShards when there are
+		// multiple SQLs
+		uniqueShards := map[string]*ShardResult{}
+		for _, result := range execResult.SuccessShards {
+			uniqueShards[result.Shard] = &result
+		}
+		var wg sync.WaitGroup
+		// If all shards succeeded, wait (up to waitReplicasTimeout) for replicas to
+		// execute the schema change via replication. This is best-effort, meaning
+		// we still return overall success if the timeout expires.
+		concurrency := sync2.NewSemaphore(10, 0)
+		reloadCtx, cancel := context.WithTimeout(ctx, exec.waitReplicasTimeout)
+		defer cancel()
+		for _, result := range uniqueShards {
+			wg.Add(1)
+			go func(result *ShardResult) {
+				defer wg.Done()
+				exec.logger.Infof("ReloadSchema on shard: %s", result.Shard)
+				schematools.ReloadShard(
+					reloadCtx,
+					exec.ts,
+					exec.tmc,
+					exec.logger,
+					exec.keyspace,
+					result.Shard,
+					result.Position,
+					concurrency,
+					false, /* includePrimary */
+				)
+			}(result)
+		}
+		wg.Wait()
+	} else {
+		exec.logger.Infof("Skipped ReloadSchema since all SQLs executed asynchronously")
+	}
 	return &execResult
 }
 
@@ -401,31 +443,6 @@ func (exec *TabletExecutor) executeOnAllTablets(ctx context.Context, execResult
 	if len(execResult.FailedShards) > 0 {
 		return
 	}
-
-	// If all shards succeeded, wait (up to waitReplicasTimeout) for replicas to
-	// execute the schema change via replication. This is best-effort, meaning
-	// we still return overall success if the timeout expires.
-	concurrency := sync2.NewSemaphore(10, 0)
-	reloadCtx, cancel := context.WithTimeout(ctx, exec.waitReplicasTimeout)
-	defer cancel()
-	for _, result := range execResult.SuccessShards {
-		wg.Add(1)
-		go func(result ShardResult) {
-			defer wg.Done()
-			schematools.ReloadShard(
-				reloadCtx,
-				exec.ts,
-				exec.tmc,
-				exec.logger,
-				exec.keyspace,
-				result.Shard,
-				result.Position,
-				concurrency,
-				false, /* includePrimary */
-			)
-		}(result)
-	}
-	wg.Wait()
 }
 
 func (exec *TabletExecutor) executeOneTablet(
diff --git a/go/vt/vtgate/engine/cached_size.go b/go/vt/vtgate/engine/cached_size.go
index 323e48bb4d..d5e7973864 100644
--- a/go/vt/vtgate/engine/cached_size.go
+++ b/go/vt/vtgate/engine/cached_size.go
@@ -154,7 +154,9 @@ func (cached *DML) CachedSize(alloc bool) int64 {
 		size += cc.CachedSize(true)
 	}
 	// field Table *vitess.io/vitess/go/vt/vtgate/vindexes.Table
-	size += cached.Table.CachedSize(true)
+	for _, table := range cached.Table {
+		size += table.CachedSize(true)
+	}
 	// field OwnedVindexQuery string
 	size += hack.RuntimeAllocSize(int64(len(cached.OwnedVindexQuery)))
 	// field RoutingParameters *vitess.io/vitess/go/vt/vtgate/engine.RoutingParameters
diff --git a/go/vt/vtgate/engine/delete.go b/go/vt/vtgate/engine/delete.go
index a49099fa53..4a8257741c 100644
--- a/go/vt/vtgate/engine/delete.go
+++ b/go/vt/vtgate/engine/delete.go
@@ -42,24 +42,6 @@ type Delete struct {
 	noInputs
 }
 
-// RouteType returns a description of the query routing type used by the primitive
-func (del *Delete) RouteType() string {
-	return del.Opcode.String()
-}
-
-// GetKeyspaceName specifies the Keyspace that this primitive routes to.
-func (del *Delete) GetKeyspaceName() string {
-	return del.Keyspace.Name
-}
-
-// GetTableName specifies the table that this primitive routes to.
-func (del *Delete) GetTableName() string {
-	if del.Table != nil {
-		return del.Table.Name.String()
-	}
-	return ""
-}
-
 // TryExecute performs a non-streaming exec.
 func (del *Delete) TryExecute(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable, _ bool) (*sqltypes.Result, error) {
 	if del.QueryTimeout != 0 {
@@ -130,7 +112,11 @@ func (del *Delete) deleteVindexEntries(ctx context.Context, vcursor VCursor, bin
 			return err
 		}
 		colnum := del.KsidLength
-		for _, colVindex := range del.Table.Owned {
+		vindexTable, err := del.GetSingleTable()
+		if err != nil {
+			return err
+		}
+		for _, colVindex := range vindexTable.Owned {
 			// Fetch the column values. colnum must keep incrementing.
 			fromIds := make([]sqltypes.Value, 0, len(colVindex.Columns))
 			for range colVindex.Columns {
diff --git a/go/vt/vtgate/engine/dml.go b/go/vt/vtgate/engine/dml.go
index 5862f26016..990b5a0a4e 100644
--- a/go/vt/vtgate/engine/dml.go
+++ b/go/vt/vtgate/engine/dml.go
@@ -19,6 +19,8 @@ package engine
 import (
 	"context"
 	"fmt"
+	"sort"
+	"strings"
 
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/key"
@@ -44,7 +46,7 @@ type DML struct {
 	KsidLength int
 
 	// Table specifies the table for the update.
-	Table *vindexes.Table
+	Table []*vindexes.Table
 
 	// OwnedVindexQuery is used for updating changes in lookup vindexes.
 	OwnedVindexQuery string
@@ -89,6 +91,43 @@ func (dml *DML) execMultiDestination(ctx context.Context, vcursor VCursor, bindV
 	return execMultiShard(ctx, vcursor, rss, queries, dml.MultiShardAutocommit)
 }
 
+// RouteType returns a description of the query routing type used by the primitive
+func (dml *DML) RouteType() string {
+	return dml.Opcode.String()
+}
+
+// GetKeyspaceName specifies the Keyspace that this primitive routes to.
+func (dml *DML) GetKeyspaceName() string {
+	return dml.Keyspace.Name
+}
+
+// GetTableName specifies the table that this primitive routes to.
+func (dml *DML) GetTableName() string {
+	if dml.Table != nil {
+		tableNameMap := map[string]any{}
+		for _, table := range dml.Table {
+			tableNameMap[table.Name.String()] = nil
+		}
+
+		var tableNames []string
+		for name := range tableNameMap {
+			tableNames = append(tableNames, name)
+		}
+		sort.Strings(tableNames)
+
+		return strings.Join(tableNames, ", ")
+	}
+	return ""
+}
+
+// GetSingleTable returns single table used in dml.
+func (dml *DML) GetSingleTable() (*vindexes.Table, error) {
+	if len(dml.Table) > 1 {
+		return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported dml on complex table expression")
+	}
+	return dml.Table[0], nil
+}
+
 func allowOnlyPrimary(rss ...*srvtopo.ResolvedShard) error {
 	for _, rs := range rss {
 		if rs != nil && rs.Target.TabletType != topodatapb.TabletType_PRIMARY {
diff --git a/go/vt/vtgate/engine/update.go b/go/vt/vtgate/engine/update.go
index db88405542..af9123253e 100644
--- a/go/vt/vtgate/engine/update.go
+++ b/go/vt/vtgate/engine/update.go
@@ -52,24 +52,6 @@ type Update struct {
 	noInputs
 }
 
-// RouteType returns a description of the query routing type used by the primitive
-func (upd *Update) RouteType() string {
-	return upd.Opcode.String()
-}
-
-// GetKeyspaceName specifies the Keyspace that this primitive routes to.
-func (upd *Update) GetKeyspaceName() string {
-	return upd.Keyspace.Name
-}
-
-// GetTableName specifies the table that this primitive routes to.
-func (upd *Update) GetTableName() string {
-	if upd.Table != nil {
-		return upd.Table.Name.String()
-	}
-	return ""
-}
-
 // TryExecute performs a non-streaming exec.
 func (upd *Update) TryExecute(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool) (*sqltypes.Result, error) {
 	if upd.QueryTimeout != 0 {
@@ -150,7 +132,11 @@ func (upd *Update) updateVindexEntries(ctx context.Context, vcursor VCursor, bin
 			return err
 		}
 
-		for _, colVindex := range upd.Table.ColumnVindexes {
+		vindexTable, err := upd.GetSingleTable()
+		if err != nil {
+			return err
+		}
+		for _, colVindex := range vindexTable.ColumnVindexes {
 			// Skip this vindex if no rows are being changed
 			updColValues, ok := upd.ChangedVindexValues[colVindex.Name]
 			if !ok {
diff --git a/go/vt/vtgate/planbuilder/delete.go b/go/vt/vtgate/planbuilder/delete.go
index 9475c17c45..62148520db 100644
--- a/go/vt/vtgate/planbuilder/delete.go
+++ b/go/vt/vtgate/planbuilder/delete.go
@@ -51,17 +51,21 @@ func buildDeletePlan(stmt sqlparser.Statement, reservedVars *sqlparser.ReservedV
 		return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "multi-table delete statement in not supported in sharded database")
 	}
 
-	if len(del.Targets) == 1 && del.Targets[0].Name != edel.Table.Name {
+	edelTable, err := edel.GetSingleTable()
+	if err != nil {
+		return nil, err
+	}
+	if len(del.Targets) == 1 && del.Targets[0].Name != edelTable.Name {
 		return nil, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.UnknownTable, "Unknown table '%s' in MULTI DELETE", del.Targets[0].Name.String())
 	}
 
-	if len(edel.Table.Owned) > 0 {
+	if len(edelTable.Owned) > 0 {
 		aTblExpr, ok := del.TableExprs[0].(*sqlparser.AliasedTableExpr)
 		if !ok {
 			return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: delete on complex table expression")
 		}
-		tblExpr := &sqlparser.AliasedTableExpr{Expr: sqlparser.TableName{Name: edel.Table.Name}, As: aTblExpr.As}
-		edel.OwnedVindexQuery = generateDMLSubquery(tblExpr, del.Where, del.OrderBy, del.Limit, edel.Table, ksidVindex.Columns)
+		tblExpr := &sqlparser.AliasedTableExpr{Expr: sqlparser.TableName{Name: edelTable.Name}, As: aTblExpr.As}
+		edel.OwnedVindexQuery = generateDMLSubquery(tblExpr, del.Where, del.OrderBy, del.Limit, edelTable, ksidVindex.Columns)
 		edel.KsidVindex = ksidVindex.Vindex
 		edel.KsidLength = len(ksidVindex.Columns)
 	}
diff --git a/go/vt/vtgate/planbuilder/dml_planner.go b/go/vt/vtgate/planbuilder/dml_planner.go
index 818be02d67..b1d05ee642 100644
--- a/go/vt/vtgate/planbuilder/dml_planner.go
+++ b/go/vt/vtgate/planbuilder/dml_planner.go
@@ -285,12 +285,17 @@ func buildDMLPlan(vschema plancontext.VSchema, dmlType string, stmt sqlparser.St
 		return nil, nil, err
 	}
 	edml.Keyspace = rb.eroute.Keyspace
+	edml.Table, err = pb.st.AllVschemaTableNames()
+	if err != nil {
+		return nil, nil, err
+	}
 	if !edml.Keyspace.Sharded {
 		// We only validate non-table subexpressions because the previous analysis has already validated them.
 		var subqueryArgs []sqlparser.SQLNode
 		subqueryArgs = append(subqueryArgs, nodes...)
 		subqueryArgs = append(subqueryArgs, where, orderBy, limit)
-		if pb.finalizeUnshardedDMLSubqueries(reservedVars, subqueryArgs...) {
+		subqueryIsUnsharded, subqueryTables := pb.finalizeUnshardedDMLSubqueries(reservedVars, subqueryArgs...)
+		if subqueryIsUnsharded {
 			vschema.WarnUnshardedOnly("subqueries can't be sharded in DML")
 		} else {
 			return nil, nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: sharded subqueries in DML")
@@ -299,6 +304,7 @@ func buildDMLPlan(vschema plancontext.VSchema, dmlType string, stmt sqlparser.St
 		// Generate query after all the analysis. Otherwise table name substitutions for
 		// routed tables won't happen.
 		edml.Query = generateQuery(stmt)
+		edml.Table = append(edml.Table, subqueryTables...)
 		return edml, nil, nil
 	}
 
@@ -320,12 +326,11 @@ func buildDMLPlan(vschema plancontext.VSchema, dmlType string, stmt sqlparser.St
 	if len(pb.st.tables) != 1 {
 		return nil, nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "multi-table %s statement is not supported in sharded database", dmlType)
 	}
-	for _, tval := range pb.st.tables {
-		// There is only one table.
-		edml.Table = tval.vschemaTable
+	edmlTable, err := edml.GetSingleTable()
+	if err != nil {
+		return nil, nil, err
 	}
-
-	routingType, ksidVindex, vindex, values, err := getDMLRouting(where, edml.Table)
+	routingType, ksidVindex, vindex, values, err := getDMLRouting(where, edmlTable)
 	if err != nil {
 		return nil, nil, err
 	}
diff --git a/go/vt/vtgate/planbuilder/expr.go b/go/vt/vtgate/planbuilder/expr.go
index 7ba2588cfe..72e45b660b 100644
--- a/go/vt/vtgate/planbuilder/expr.go
+++ b/go/vt/vtgate/planbuilder/expr.go
@@ -24,6 +24,7 @@ import (
 
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vtgate/engine"
+	"vitess.io/vitess/go/vt/vtgate/vindexes"
 )
 
 type subqueryInfo struct {
@@ -209,13 +210,14 @@ func hasSubquery(node sqlparser.SQLNode) bool {
 	return has
 }
 
-func (pb *primitiveBuilder) finalizeUnshardedDMLSubqueries(reservedVars *sqlparser.ReservedVars, nodes ...sqlparser.SQLNode) bool {
+func (pb *primitiveBuilder) finalizeUnshardedDMLSubqueries(reservedVars *sqlparser.ReservedVars, nodes ...sqlparser.SQLNode) (bool, []*vindexes.Table) {
 	var keyspace string
+	var tables []*vindexes.Table
 	if rb, ok := pb.plan.(*route); ok {
 		keyspace = rb.eroute.Keyspace.Name
 	} else {
 		// This code is unreachable because the caller checks.
-		return false
+		return false, nil
 	}
 
 	for _, node := range nodes {
@@ -247,6 +249,11 @@ func (pb *primitiveBuilder) finalizeUnshardedDMLSubqueries(reservedVars *sqlpars
 				for _, sub := range innerRoute.substitutions {
 					*sub.oldExpr = *sub.newExpr
 				}
+				spbTables, err := spb.st.AllVschemaTableNames()
+				if err != nil {
+					return false, err
+				}
+				tables = append(tables, spbTables...)
 			case *sqlparser.Union:
 				if !inSubQuery {
 					return true, nil
@@ -270,10 +277,10 @@ func (pb *primitiveBuilder) finalizeUnshardedDMLSubqueries(reservedVars *sqlpars
 			return true, nil
 		}, node)
 		if !samePlan {
-			return false
+			return false, nil
 		}
 	}
-	return true
+	return true, tables
 }
 
 func valEqual(a, b sqlparser.Expr) bool {
diff --git a/go/vt/vtgate/planbuilder/from.go b/go/vt/vtgate/planbuilder/from.go
index d6976a6cd2..18d653778a 100644
--- a/go/vt/vtgate/planbuilder/from.go
+++ b/go/vt/vtgate/planbuilder/from.go
@@ -19,6 +19,7 @@ package planbuilder
 import (
 	"errors"
 	"fmt"
+	"sort"
 	"strings"
 
 	"vitess.io/vitess/go/mysql/collations"
@@ -153,8 +154,18 @@ func (pb *primitiveBuilder) processAliasedTable(tableExpr *sqlparser.AliasedTabl
 		// a new set of column references will be generated against the new tables,
 		// and those vindex maps will be returned. They have to replace the old vindex
 		// maps of the inherited route options.
+		var tableNames []string
+		spbTables, err := spb.st.AllVschemaTableNames()
+		if err != nil {
+			return err
+		}
+		for _, table := range spbTables {
+			tableNames = append(tableNames, table.Name.String())
+		}
+		sort.Strings(tableNames)
 		vschemaTable := &vindexes.Table{
 			Keyspace: subroute.eroute.Keyspace,
+			Name:     sqlparser.NewIdentifierCS(strings.Join(tableNames, ", ")),
 		}
 		for _, rc := range subroute.ResultColumns() {
 			if rc.column.vindex == nil {
diff --git a/go/vt/vtgate/planbuilder/gen4_compare_v3_planner.go b/go/vt/vtgate/planbuilder/gen4_compare_v3_planner.go
index 10135699f9..08c43e2e1b 100644
--- a/go/vt/vtgate/planbuilder/gen4_compare_v3_planner.go
+++ b/go/vt/vtgate/planbuilder/gen4_compare_v3_planner.go
@@ -25,6 +25,11 @@ import (
 
 func gen4CompareV3Planner(query string) func(sqlparser.Statement, *sqlparser.ReservedVars, plancontext.VSchema) (engine.Primitive, error) {
 	return func(statement sqlparser.Statement, vars *sqlparser.ReservedVars, ctxVSchema plancontext.VSchema) (engine.Primitive, error) {
+		// we will be switching the planner version to Gen4 and V3 in order to
+		// create instructions using them, thus we make sure to switch back to
+		// the Gen4CompareV3 planner before exiting this method.
+		defer ctxVSchema.SetPlannerVersion(Gen4CompareV3)
+
 		switch statement.(type) {
 		case *sqlparser.Select, *sqlparser.Union:
 		// These we can compare. Everything else we'll just use the Gen4 planner
@@ -32,11 +37,6 @@ func gen4CompareV3Planner(query string) func(sqlparser.Statement, *sqlparser.Res
 			return planWithPlannerVersion(statement, vars, ctxVSchema, query, Gen4)
 		}
 
-		// we will be switching the planner version to Gen4 and V3 in order to
-		// create instructions using them, thus we make sure to switch back to
-		// the Gen4CompareV3 planner before exiting this method.
-		defer ctxVSchema.SetPlannerVersion(Gen4CompareV3)
-
 		// preliminary checks on the given statement
 		onlyGen4, hasOrderBy, err := preliminaryChecks(statement)
 		if err != nil {
diff --git a/go/vt/vtgate/planbuilder/gen4_planner.go b/go/vt/vtgate/planbuilder/gen4_planner.go
index 3559e34069..a062e4f42c 100644
--- a/go/vt/vtgate/planbuilder/gen4_planner.go
+++ b/go/vt/vtgate/planbuilder/gen4_planner.go
@@ -170,7 +170,7 @@ func newBuildSelectPlan(
 	// record any warning as planner warning.
 	vschema.PlannerWarning(semTable.Warning)
 
-	if ks := semTable.SingleUnshardedKeyspace(); ks != nil {
+	if ks, _ := semTable.SingleUnshardedKeyspace(); ks != nil {
 		return unshardedShortcut(selStmt, ks, semTable)
 	}
 
@@ -254,9 +254,10 @@ func gen4UpdateStmtPlanner(
 		return nil, err
 	}
 
-	if ks := semTable.SingleUnshardedKeyspace(); ks != nil {
+	if ks, tables := semTable.SingleUnshardedKeyspace(); ks != nil {
 		edml := engine.NewDML()
 		edml.Keyspace = ks
+		edml.Table = tables
 		edml.Opcode = engine.Unsharded
 		edml.Query = generateQuery(updStmt)
 		upd := &engine.Update{DML: edml}
diff --git a/go/vt/vtgate/planbuilder/operator_transformers.go b/go/vt/vtgate/planbuilder/operator_transformers.go
index a66ec558d6..9bc0fd986b 100644
--- a/go/vt/vtgate/planbuilder/operator_transformers.go
+++ b/go/vt/vtgate/planbuilder/operator_transformers.go
@@ -180,8 +180,10 @@ func transformUpdatePlan(ctx *plancontext.PlanningContext, op *physical.Route, u
 	ast := upd.AST
 	replaceSubQuery(ctx, ast)
 	edml := &engine.DML{
-		Query:            generateQuery(ast),
-		Table:            upd.VTable,
+		Query: generateQuery(ast),
+		Table: []*vindexes.Table{
+			upd.VTable,
+		},
 		OwnedVindexQuery: upd.OwnedVindexQuery,
 		RoutingParameters: &engine.RoutingParameters{
 			Opcode:            op.RouteOpCode,
diff --git a/go/vt/vtgate/planbuilder/physical/route_planning.go b/go/vt/vtgate/planbuilder/physical/route_planning.go
index e163529575..9e0d86f489 100644
--- a/go/vt/vtgate/planbuilder/physical/route_planning.go
+++ b/go/vt/vtgate/planbuilder/physical/route_planning.go
@@ -1092,6 +1092,11 @@ func pushJoinPredicateOnJoin(ctx *plancontext.PlanningContext, exprs []sqlparser
 
 	node.LHS = lhsPlan
 	node.RHS = rhsPlan
+	// If the predicate field is previously non-empty
+	// keep that predicate too
+	if node.Predicate != nil {
+		exprs = append(exprs, node.Predicate)
+	}
 	node.Predicate = sqlparser.AndExpressions(exprs...)
 	return node, nil
 }
diff --git a/go/vt/vtgate/planbuilder/symtab.go b/go/vt/vtgate/planbuilder/symtab.go
index 395b45c5f9..f3f3bf6997 100644
--- a/go/vt/vtgate/planbuilder/symtab.go
+++ b/go/vt/vtgate/planbuilder/symtab.go
@@ -205,6 +205,24 @@ func (st *symtab) AllTables() []*table {
 	return tables
 }
 
+// AllVschemaTableNames returns an ordered list of all current vschema tables.
+func (st *symtab) AllVschemaTableNames() ([]*vindexes.Table, error) {
+	if len(st.tableNames) == 0 {
+		return nil, nil
+	}
+	tables := make([]*vindexes.Table, 0, len(st.tableNames))
+	for _, tname := range st.tableNames {
+		t, ok := st.tables[tname]
+		if !ok {
+			return nil, fmt.Errorf("table %v not found", sqlparser.String(tname))
+		}
+		if t.vschemaTable != nil {
+			tables = append(tables, t.vschemaTable)
+		}
+	}
+	return tables, nil
+}
+
 // FindTable finds a table in symtab. This function is specifically used
 // for expanding 'select a.*' constructs. If you're in a subquery,
 // you're most likely referring to a table in the local 'from' clause.
diff --git a/go/vt/vtgate/planbuilder/update_planner.go b/go/vt/vtgate/planbuilder/update_planner.go
index c295cdc49a..21a5be60b6 100644
--- a/go/vt/vtgate/planbuilder/update_planner.go
+++ b/go/vt/vtgate/planbuilder/update_planner.go
@@ -45,8 +45,11 @@ func buildUpdatePlan(query string) stmtPlanner {
 		if dml.Opcode == engine.Unsharded {
 			return eupd, nil
 		}
-
-		cvv, ovq, err := buildChangedVindexesValues(upd, eupd.Table, ksidVindex.Columns)
+		eupdTable, err := eupd.GetSingleTable()
+		if err != nil {
+			return nil, err
+		}
+		cvv, ovq, err := buildChangedVindexesValues(upd, eupdTable, ksidVindex.Columns)
 		if err != nil {
 			return nil, err
 		}
diff --git a/go/vt/vtgate/semantics/semantic_state.go b/go/vt/vtgate/semantics/semantic_state.go
index 4fe0ce7537..9592556cd4 100644
--- a/go/vt/vtgate/semantics/semantic_state.go
+++ b/go/vt/vtgate/semantics/semantic_state.go
@@ -366,10 +366,12 @@ func (st *SemTable) ColumnLookup(col *sqlparser.ColName) (int, error) {
 }
 
 // SingleUnshardedKeyspace returns the single keyspace if all tables in the query are in the same, unsharded keyspace
-func (st *SemTable) SingleUnshardedKeyspace() *vindexes.Keyspace {
+func (st *SemTable) SingleUnshardedKeyspace() (*vindexes.Keyspace, []*vindexes.Table) {
 	var ks *vindexes.Keyspace
+	var tables []*vindexes.Table
 	for _, table := range st.Tables {
 		vindexTable := table.GetVindexTable()
+
 		if vindexTable == nil || vindexTable.Type != "" {
 			_, isDT := table.getExpr().Expr.(*sqlparser.DerivedTable)
 			if isDT {
@@ -377,27 +379,28 @@ func (st *SemTable) SingleUnshardedKeyspace() *vindexes.Keyspace {
 				// we check the real tables inside the derived table as well for same unsharded keyspace.
 				continue
 			}
-			return nil
+			return nil, nil
 		}
 		name, ok := table.getExpr().Expr.(sqlparser.TableName)
 		if !ok {
-			return nil
+			return nil, nil
 		}
 		if name.Name.String() != vindexTable.Name.String() {
 			// this points to a table alias. safer to not shortcut
-			return nil
+			return nil, nil
 		}
 		this := vindexTable.Keyspace
 		if this == nil || this.Sharded {
-			return nil
+			return nil, nil
 		}
 		if ks == nil {
 			ks = this
 		} else {
 			if ks != this {
-				return nil
+				return nil, nil
 			}
 		}
+		tables = append(tables, vindexTable)
 	}
-	return ks
+	return ks, tables
 }


Test patch:
diff --git a/go/mysql/collations/integration/collations_test.go b/go/mysql/collations/integration/collations_test.go
index 89dd4ace2a..e91719090b 100644
--- a/go/mysql/collations/integration/collations_test.go
+++ b/go/mysql/collations/integration/collations_test.go
@@ -35,9 +35,19 @@ import (
 	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/mysql/collations/remote"
 	"vitess.io/vitess/go/sqltypes"
+	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/sqlparser"
 )
 
+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
+	collationEnv = collations.NewEnvironment(mySQLVersion)
+}
+
 func getSQLQueries(t *testing.T, testfile string) []string {
 	tf, err := os.Open(testfile)
 	if err != nil {
@@ -105,7 +115,7 @@ func parseWeightString(b []byte) []byte {
 }
 
 func (u *uca900CollationTest) Test(t *testing.T, result *sqltypes.Result) {
-	coll := collations.Local().LookupByName(u.collation)
+	coll := collationEnv.LookupByName(u.collation)
 	if coll == nil {
 		t.Fatalf("unknown collation %q", u.collation)
 	}
@@ -215,7 +225,7 @@ func TestCollationWithSpace(t *testing.T) {
 
 	for _, collName := range []string{"utf8mb4_0900_ai_ci", "utf8mb4_unicode_ci", "utf8mb4_unicode_520_ci"} {
 		t.Run(collName, func(t *testing.T) {
-			local := collations.Local().LookupByName(collName)
+			local := collationEnv.LookupByName(collName)
 			remote := remote.NewCollation(conn, collName)
 
 			for _, size := range []int{0, codepoints, codepoints + 1, codepoints + 2, 20, 32} {
diff --git a/go/mysql/endtoend/query_test.go b/go/mysql/endtoend/query_test.go
index 1118c46f85..f6410147a7 100644
--- a/go/mysql/endtoend/query_test.go
+++ b/go/mysql/endtoend/query_test.go
@@ -33,6 +33,10 @@ import (
 	querypb "vitess.io/vitess/go/vt/proto/query"
 )
 
+const (
+	charsetName = "utf8mb4"
+)
+
 func columnSize(cs collations.ID, size uint32) uint32 {
 	// utf8_general_ci results in smaller max column sizes because MySQL 5.7 is silly
 	if collations.Local().LookupByID(cs).Charset().Name() == "utf8mb3" {
@@ -78,6 +82,7 @@ func TestQueries(t *testing.T) {
 	if err != nil {
 		t.Fatalf("insert failed: %v", err)
 	}
+	collID := getDefaultCollationID()
 	expectedResult := &sqltypes.Result{
 		Fields: []*querypb.Field{
 			{
@@ -101,8 +106,8 @@ func TestQueries(t *testing.T) {
 				OrgTable:     "a",
 				Database:     "vttest",
 				OrgName:      "name",
-				ColumnLength: columnSize(conn.CharacterSet, 512),
-				Charset:      uint32(conn.CharacterSet),
+				ColumnLength: columnSize(collID, 512),
+				Charset:      uint32(collID),
 			},
 		},
 		Rows: [][]sqltypes.Value{
@@ -188,6 +193,7 @@ func readRowsUsingStream(t *testing.T, conn *mysql.Conn, expectedCount int) {
 	}
 
 	// Check the fields.
+	collID := getDefaultCollationID()
 	expectedFields := []*querypb.Field{
 		{
 			Name:         "id",
@@ -210,8 +216,8 @@ func readRowsUsingStream(t *testing.T, conn *mysql.Conn, expectedCount int) {
 			OrgTable:     "a",
 			Database:     "vttest",
 			OrgName:      "name",
-			ColumnLength: columnSize(conn.CharacterSet, 512),
-			Charset:      uint32(conn.CharacterSet),
+			ColumnLength: columnSize(collID, 512),
+			Charset:      uint32(collID),
 		},
 	}
 	fields, err := conn.Fields()
@@ -306,7 +312,8 @@ func TestSysInfo(t *testing.T) {
 	_, err = conn.ExecuteFetch("drop table if exists `a`", 1000, true)
 	require.NoError(t, err)
 
-	_, err = conn.ExecuteFetch("CREATE TABLE `a` (`one` int NOT NULL,`two` int NOT NULL,PRIMARY KEY (`one`,`two`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4", 1000, true)
+	_, err = conn.ExecuteFetch(fmt.Sprintf("CREATE TABLE `a` (`one` int NOT NULL,`two` int NOT NULL,PRIMARY KEY (`one`,`two`)) ENGINE=InnoDB DEFAULT CHARSET=%s",
+		charsetName), 1000, true)
 	require.NoError(t, err)
 	defer conn.ExecuteFetch("drop table `a`", 1000, true)
 
@@ -339,3 +346,9 @@ func TestSysInfo(t *testing.T) {
 	assert.EqualValues(t, sqltypes.Uint64, qr.Fields[4].Type)
 	assert.EqualValues(t, querypb.Type_UINT64, qr.Rows[0][4].Type())
 }
+
+func getDefaultCollationID() collations.ID {
+	collationHandler := collations.Local()
+	collation := collationHandler.DefaultCollationForCharset(charsetName)
+	return collation.ID()
+}
diff --git a/go/test/endtoend/cluster/vtgate_process.go b/go/test/endtoend/cluster/vtgate_process.go
index d460c387bb..bfba50e3c1 100644
--- a/go/test/endtoend/cluster/vtgate_process.go
+++ b/go/test/endtoend/cluster/vtgate_process.go
@@ -32,6 +32,7 @@ import (
 	"time"
 
 	"vitess.io/vitess/go/vt/log"
+	"vitess.io/vitess/go/vt/mysqlctl"
 	"vitess.io/vitess/go/vt/vtgate/planbuilder"
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
 )
@@ -69,7 +70,6 @@ const defaultVtGatePlannerVersion = planbuilder.Gen4CompareV3
 
 // Setup starts Vtgate process with required arguements
 func (vtgate *VtgateProcess) Setup() (err error) {
-
 	args := []string{
 		"--topo_implementation", vtgate.CommonArg.TopoImplementation,
 		"--topo_global_server_address", vtgate.CommonArg.TopoGlobalAddress,
@@ -86,6 +86,28 @@ func (vtgate *VtgateProcess) Setup() (err error) {
 		"--service_map", vtgate.ServiceMap,
 		"--mysql_auth_server_impl", vtgate.MySQLAuthServerImpl,
 	}
+	// If no explicit mysql_server_version has been specified then we autodetect
+	// the MySQL version that will be used for the test and base the vtgate's
+	// mysql server version on that.
+	msvflag := false
+	for _, f := range vtgate.ExtraArgs {
+		if strings.Contains(f, "mysql_server_version") {
+			msvflag = true
+			break
+		}
+	}
+	if !msvflag {
+		version, err := mysqlctl.GetVersionString()
+		if err != nil {
+			return err
+		}
+		_, vers, err := mysqlctl.ParseVersionString(version)
+		if err != nil {
+			return err
+		}
+		mysqlvers := fmt.Sprintf("%d.%d.%d-vitess", vers.Major, vers.Minor, vers.Patch)
+		args = append(args, "--mysql_server_version", mysqlvers)
+	}
 	if vtgate.PlannerVersion > 0 {
 		args = append(args, "--planner-version", vtgate.PlannerVersion.String())
 	}
diff --git a/go/test/endtoend/reparent/emergencyreparent/ers_test.go b/go/test/endtoend/reparent/emergencyreparent/ers_test.go
index e0f3e1a073..9bff176a9e 100644
--- a/go/test/endtoend/reparent/emergencyreparent/ers_test.go
+++ b/go/test/endtoend/reparent/emergencyreparent/ers_test.go
@@ -22,7 +22,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"vitess.io/vitess/go/test/endtoend/cluster"
@@ -438,10 +437,7 @@ func TestERSForInitialization(t *testing.T) {
 	utils.RunSQL(context.Background(), t, "create table vt_insert_test (id bigint, msg varchar(64), primary key (id)) Engine=InnoDB", tablets[0])
 	utils.CheckPrimaryTablet(t, clusterInstance, tablets[0])
 	utils.ValidateTopology(t, clusterInstance, false)
-	time.Sleep(100 * time.Millisecond) // wait for replication to catchup
-	strArray := utils.GetShardReplicationPositions(t, clusterInstance, utils.KeyspaceName, utils.ShardName, true)
-	assert.Equal(t, len(tablets), len(strArray))
-	assert.Contains(t, strArray[0], "primary") // primary first
+	utils.WaitForReplicationToStart(t, clusterInstance, utils.KeyspaceName, utils.ShardName, len(tablets), true)
 	utils.ConfirmReplication(t, tablets[0], tablets[1:])
 }
 
diff --git a/go/test/endtoend/reparent/plannedreparent/reparent_test.go b/go/test/endtoend/reparent/plannedreparent/reparent_test.go
index 71330e0e1c..2e6b67c68d 100644
--- a/go/test/endtoend/reparent/plannedreparent/reparent_test.go
+++ b/go/test/endtoend/reparent/plannedreparent/reparent_test.go
@@ -65,9 +65,7 @@ func TestReparentGraceful(t *testing.T) {
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
 	// Run this to make sure it succeeds.
-	strArray := utils.GetShardReplicationPositions(t, clusterInstance, utils.KeyspaceName, utils.ShardName, false)
-	assert.Equal(t, 4, len(strArray))          // one primary, three replicas
-	assert.Contains(t, strArray[0], "primary") // primary first
+	utils.WaitForReplicationToStart(t, clusterInstance, utils.KeyspaceName, utils.ShardName, len(tablets), true)
 
 	// Perform a graceful reparent operation
 	utils.Prs(t, clusterInstance, tablets[1])
diff --git a/go/test/endtoend/reparent/utils/utils.go b/go/test/endtoend/reparent/utils/utils.go
index 255a663b74..bffcad957d 100644
--- a/go/test/endtoend/reparent/utils/utils.go
+++ b/go/test/endtoend/reparent/utils/utils.go
@@ -58,10 +58,11 @@ var (
 	primary key (id)
 	) Engine=InnoDB	
 `
-	cell1         = "zone1"
-	cell2         = "zone2"
-	ShardName     = "0"
-	KeyspaceShard = KeyspaceName + "/" + ShardName
+	cell1                  = "zone1"
+	cell2                  = "zone2"
+	ShardName              = "0"
+	KeyspaceShard          = KeyspaceName + "/" + ShardName
+	replicationWaitTimeout = time.Duration(15 * time.Second)
 )
 
 //region cluster setup/teardown
@@ -198,10 +199,7 @@ func setupShard(ctx context.Context, t *testing.T, clusterInstance *cluster.Loca
 	CheckPrimaryTablet(t, clusterInstance, tablets[0])
 
 	ValidateTopology(t, clusterInstance, false)
-	time.Sleep(100 * time.Millisecond) // wait for replication to catchup
-	strArray := GetShardReplicationPositions(t, clusterInstance, KeyspaceName, shardName, true)
-	assert.Equal(t, len(tablets), len(strArray))
-	assert.Contains(t, strArray[0], "primary") // primary first
+	WaitForReplicationToStart(t, clusterInstance, KeyspaceName, shardName, len(tablets), true)
 }
 
 func setupClusterLegacy(ctx context.Context, t *testing.T, shardName string, cells []string, numTablets []int, enableSemiSync bool) *cluster.LocalProcessCluster {
@@ -321,10 +319,7 @@ func setupShardLegacy(ctx context.Context, t *testing.T, clusterInstance *cluste
 	CheckPrimaryTablet(t, clusterInstance, tablets[0])
 
 	ValidateTopology(t, clusterInstance, false)
-	time.Sleep(100 * time.Millisecond) // wait for replication to catchup
-	strArray := GetShardReplicationPositions(t, clusterInstance, KeyspaceName, shardName, true)
-	assert.Equal(t, len(tablets), len(strArray))
-	assert.Contains(t, strArray[0], "primary") // primary first
+	WaitForReplicationToStart(t, clusterInstance, KeyspaceName, shardName, len(tablets), true)
 }
 
 //endregion
@@ -455,7 +450,6 @@ func ConfirmReplication(t *testing.T, primary *cluster.Vttablet, replicas []*clu
 	// insert data into the new primary, check the connected replica work
 	insertSQL := fmt.Sprintf(insertSQL, n, n)
 	RunSQL(ctx, t, insertSQL, primary)
-	time.Sleep(100 * time.Millisecond)
 	for _, tab := range replicas {
 		err := CheckInsertedValues(ctx, t, tab, n)
 		require.NoError(t, err)
@@ -518,20 +512,28 @@ func isHealthyPrimaryTablet(t *testing.T, clusterInstance *cluster.LocalProcessC
 
 // CheckInsertedValues checks that the given value is present in the given tablet
 func CheckInsertedValues(ctx context.Context, t *testing.T, tablet *cluster.Vttablet, index int) error {
+	query := fmt.Sprintf("select msg from vt_insert_test where id=%d", index)
+	tabletParams := getMysqlConnParam(tablet)
+	conn, err := mysql.Connect(ctx, &tabletParams)
+	require.Nil(t, err)
+	defer conn.Close()
+
 	// wait until it gets the data
-	timeout := time.Now().Add(15 * time.Second)
+	timeout := time.Now().Add(replicationWaitTimeout)
 	i := 0
 	for time.Now().Before(timeout) {
-		selectSQL := fmt.Sprintf("select msg from vt_insert_test where id=%d", index)
-		qr := RunSQL(ctx, t, selectSQL, tablet)
-		if len(qr.Rows) == 1 {
+		// We'll get a mysql.ERNoSuchTable (1146) error if the CREATE TABLE has not replicated yet and
+		// it's possible that we get other ephemeral errors too, so we make the tests more robust by
+		// retrying with the timeout.
+		qr, err := conn.ExecuteFetch(query, 1, true)
+		if err == nil && len(qr.Rows) == 1 {
 			return nil
 		}
 		t := time.Duration(300 * i)
 		time.Sleep(t * time.Millisecond)
 		i++
 	}
-	return fmt.Errorf("data is not yet replicated on tablet %s", tablet.Alias)
+	return fmt.Errorf("data did not get replicated on tablet %s within the timeout of %v", tablet.Alias, replicationWaitTimeout)
 }
 
 func CheckSemiSyncSetupCorrectly(t *testing.T, tablet *cluster.Vttablet, semiSyncVal string) {
@@ -618,6 +620,7 @@ func GetNewPrimary(t *testing.T, clusterInstance *cluster.LocalProcessCluster) *
 }
 
 // GetShardReplicationPositions gets the shards replication positions.
+// This should not generally be called directly, instead use the WaitForReplicationToCatchup method.
 func GetShardReplicationPositions(t *testing.T, clusterInstance *cluster.LocalProcessCluster, keyspaceName, shardName string, doPrint bool) []string {
 	output, err := clusterInstance.VtctlclientProcess.ExecuteCommandWithOutput(
 		"ShardReplicationPositions", fmt.Sprintf("%s/%s", keyspaceName, shardName))
@@ -635,6 +638,23 @@ func GetShardReplicationPositions(t *testing.T, clusterInstance *cluster.LocalPr
 	return strArray
 }
 
+func WaitForReplicationToStart(t *testing.T, clusterInstance *cluster.LocalProcessCluster, keyspaceName, shardName string, tabletCnt int, doPrint bool) {
+	tck := time.NewTicker(500 * time.Millisecond)
+	for {
+		select {
+		case <-tck.C:
+			strArray := GetShardReplicationPositions(t, clusterInstance, KeyspaceName, shardName, true)
+			if len(strArray) == tabletCnt && strings.Contains(strArray[0], "primary") { // primary first
+				return
+			}
+		case <-time.After(replicationWaitTimeout):
+			require.FailNow(t, fmt.Sprintf("replication did not start everywhere in %s/%s within the timeout of %v",
+				keyspaceName, shardName, replicationWaitTimeout))
+			return
+		}
+	}
+}
+
 // endregion
 
 // CheckReplicaStatus checks the replication status and asserts that the replication is stopped
diff --git a/go/test/endtoend/vtgate/misc_test.go b/go/test/endtoend/vtgate/misc_test.go
index 135d297ce5..5cf14b761d 100644
--- a/go/test/endtoend/vtgate/misc_test.go
+++ b/go/test/endtoend/vtgate/misc_test.go
@@ -828,3 +828,25 @@ func TestEmptyQuery(t *testing.T) {
 	utils.AssertContainsError(t, conn, ";", "Query was empty")
 	utils.AssertIsEmpty(t, conn, "-- this is a comment")
 }
+
+// TestJoinWithMergedRouteWithPredicate checks the issue found in https://github.com/vitessio/vitess/issues/10713
+func TestJoinWithMergedRouteWithPredicate(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	utils.Exec(t, conn, "delete from t1")
+	defer utils.Exec(t, conn, "delete from t1")
+	utils.Exec(t, conn, "delete from t2")
+	defer utils.Exec(t, conn, "delete from t2")
+	utils.Exec(t, conn, "delete from t3")
+	defer utils.Exec(t, conn, "delete from t3")
+
+	utils.Exec(t, conn, "insert into t1 (id1,id2) values (1, 13)")
+	utils.Exec(t, conn, "insert into t2 (id3,id4) values (5, 10), (15, 20)")
+	utils.Exec(t, conn, "insert into t3 (id5,id6,id7) values (13, 5, 8)")
+
+	utils.AssertMatches(t, conn, "select t3.id7, t2.id3, t3.id6 from t1 join t3 on t1.id2 = t3.id5 join t2 on t3.id6 = t2.id3 where t1.id2 = 13", `[[INT64(8) INT64(5) INT64(5)]]`)
+}
diff --git a/go/test/endtoend/vtgate/reservedconn/main_test.go b/go/test/endtoend/vtgate/reservedconn/main_test.go
index 08f7f544d5..2dcd66109f 100644
--- a/go/test/endtoend/vtgate/reservedconn/main_test.go
+++ b/go/test/endtoend/vtgate/reservedconn/main_test.go
@@ -124,6 +124,8 @@ func TestMain(m *testing.M) {
 		}
 
 		// Start vtgate
+		// This test requires setting the mysql_server_version vtgate flag
+		// to 5.7 regardless of the actual MySQL version used for the tests.
 		clusterInstance.VtGateExtraArgs = []string{"--lock_heartbeat_time", "2s", "--mysql_server_version", "5.7.0"}
 		if err := clusterInstance.StartVtgate(); err != nil {
 			return 1
diff --git a/go/vt/vtgate/engine/delete_test.go b/go/vt/vtgate/engine/delete_test.go
index 3b98feba53..25c30307e3 100644
--- a/go/vt/vtgate/engine/delete_test.go
+++ b/go/vt/vtgate/engine/delete_test.go
@@ -197,8 +197,10 @@ func TestDeleteOwnedVindex(t *testing.T) {
 				Vindex:   ks.Vindexes["hash"],
 				Values:   []evalengine.Expr{evalengine.NewLiteralInt(1)},
 			},
-			Query:            "dummy_delete",
-			Table:            ks.Tables["t1"],
+			Query: "dummy_delete",
+			Table: []*vindexes.Table{
+				ks.Tables["t1"],
+			},
 			OwnedVindexQuery: "dummy_subquery",
 			KsidVindex:       ks.Vindexes["hash"],
 			KsidLength:       1,
@@ -283,8 +285,10 @@ func TestDeleteOwnedVindexMultiCol(t *testing.T) {
 				Vindex:   ks.Vindexes["rg_vdx"],
 				Values:   []evalengine.Expr{evalengine.NewLiteralInt(1), evalengine.NewLiteralInt(2)},
 			},
-			Query:            "dummy_delete",
-			Table:            ks.Tables["rg_tbl"],
+			Query: "dummy_delete",
+			Table: []*vindexes.Table{
+				ks.Tables["rg_tbl"],
+			},
 			OwnedVindexQuery: "dummy_subquery",
 			KsidVindex:       ks.Vindexes["rg_vdx"],
 			KsidLength:       2,
@@ -365,7 +369,9 @@ func TestDeleteSharded(t *testing.T) {
 				Keyspace: ks.Keyspace,
 			},
 			Query: "dummy_delete",
-			Table: ks.Tables["t2"],
+			Table: []*vindexes.Table{
+				ks.Tables["t2"],
+			},
 		},
 	}
 
@@ -392,7 +398,9 @@ func TestDeleteShardedStreaming(t *testing.T) {
 				Keyspace: ks.Keyspace,
 			},
 			Query: "dummy_delete",
-			Table: ks.Tables["t2"],
+			Table: []*vindexes.Table{
+				ks.Tables["t2"],
+			},
 		},
 	}
 
@@ -415,8 +423,10 @@ func TestDeleteScatterOwnedVindex(t *testing.T) {
 				Opcode:   Scatter,
 				Keyspace: ks.Keyspace,
 			},
-			Query:            "dummy_delete",
-			Table:            ks.Tables["t1"],
+			Query: "dummy_delete",
+			Table: []*vindexes.Table{
+				ks.Tables["t1"],
+			},
 			OwnedVindexQuery: "dummy_subquery",
 			KsidVindex:       ks.Vindexes["hash"],
 			KsidLength:       1,
@@ -505,8 +515,10 @@ func TestDeleteInChangedVindexMultiCol(t *testing.T) {
 					evalengine.NewLiteralInt(3),
 				},
 			},
-			Query:            "dummy_update",
-			Table:            ks.Tables["rg_tbl"],
+			Query: "dummy_update",
+			Table: []*vindexes.Table{
+				ks.Tables["rg_tbl"],
+			},
 			OwnedVindexQuery: "dummy_subquery",
 			KsidVindex:       ks.Vindexes["rg_vdx"],
 			KsidLength:       2,
diff --git a/go/vt/vtgate/engine/memory_sort_test.go b/go/vt/vtgate/engine/memory_sort_test.go
index f24b968e37..80b75b6177 100644
--- a/go/vt/vtgate/engine/memory_sort_test.go
+++ b/go/vt/vtgate/engine/memory_sort_test.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"testing"
 
+	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
 
 	"vitess.io/vitess/go/mysql/collations"
@@ -31,6 +32,13 @@ import (
 	querypb "vitess.io/vitess/go/vt/proto/query"
 )
 
+func init() {
+	// We require MySQL 8.0 collations for the comparisons in the tests
+	mySQLVersion := "8.0.0"
+	servenv.MySQLServerVersion = &mySQLVersion
+	collationEnv = collations.NewEnvironment(mySQLVersion)
+}
+
 func TestMemorySortExecute(t *testing.T) {
 	fields := sqltypes.MakeTestFields(
 		"c1|c2",
diff --git a/go/vt/vtgate/engine/ordered_aggregate_test.go b/go/vt/vtgate/engine/ordered_aggregate_test.go
index 6e288cf6d5..461df17796 100644
--- a/go/vt/vtgate/engine/ordered_aggregate_test.go
+++ b/go/vt/vtgate/engine/ordered_aggregate_test.go
@@ -31,8 +31,18 @@ import (
 
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
 	querypb "vitess.io/vitess/go/vt/proto/query"
+	"vitess.io/vitess/go/vt/servenv"
 )
 
+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
+	collationEnv = collations.NewEnvironment(mySQLVersion)
+}
+
 func TestOrderedAggregateExecute(t *testing.T) {
 	assert := assert.New(t)
 	fields := sqltypes.MakeTestFields(
@@ -999,7 +1009,7 @@ func TestOrderedAggregateCollate(t *testing.T) {
 		)},
 	}
 
-	collationID, _ := collations.Local().LookupID("utf8mb4_0900_ai_ci")
+	collationID, _ := collationEnv.LookupID("utf8mb4_0900_ai_ci")
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
 			Opcode: AggregateSum,
@@ -1042,7 +1052,7 @@ func TestOrderedAggregateCollateAS(t *testing.T) {
 		)},
 	}
 
-	collationID, _ := collations.Local().LookupID("utf8mb4_0900_as_ci")
+	collationID, _ := collationEnv.LookupID("utf8mb4_0900_as_ci")
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
 			Opcode: AggregateSum,
@@ -1087,7 +1097,7 @@ func TestOrderedAggregateCollateKS(t *testing.T) {
 		)},
 	}
 
-	collationID, _ := collations.Local().LookupID("utf8mb4_ja_0900_as_cs_ks")
+	collationID, _ := collationEnv.LookupID("utf8mb4_ja_0900_as_cs_ks")
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
 			Opcode: AggregateSum,
diff --git a/go/vt/vtgate/engine/route_test.go b/go/vt/vtgate/engine/route_test.go
index 569bb8d4c0..e7d58d66cd 100644
--- a/go/vt/vtgate/engine/route_test.go
+++ b/go/vt/vtgate/engine/route_test.go
@@ -27,6 +27,7 @@ import (
 
 	"vitess.io/vitess/go/mysql/collations"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/vterrors"
 
 	"vitess.io/vitess/go/vt/sqlparser"
@@ -49,6 +50,13 @@ var defaultSelectResult = sqltypes.MakeTestResult(
 	"1",
 )
 
+func init() {
+	// We require MySQL 8.0 collations for the comparisons in the tests
+	mySQLVersion := "8.0.0"
+	servenv.MySQLServerVersion = &mySQLVersion
+	collationEnv = collations.NewEnvironment(mySQLVersion)
+}
+
 func TestSelectUnsharded(t *testing.T) {
 	sel := NewRoute(
 		Unsharded,
diff --git a/go/vt/vtgate/engine/update_test.go b/go/vt/vtgate/engine/update_test.go
index 7531c70076..46bbf6a751 100644
--- a/go/vt/vtgate/engine/update_test.go
+++ b/go/vt/vtgate/engine/update_test.go
@@ -247,8 +247,10 @@ func TestUpdateEqualChangedVindex(t *testing.T) {
 				Vindex:   ks.Vindexes["hash"],
 				Values:   []evalengine.Expr{evalengine.NewLiteralInt(1)},
 			},
-			Query:            "dummy_update",
-			Table:            ks.Tables["t1"],
+			Query: "dummy_update",
+			Table: []*vindexes.Table{
+				ks.Tables["t1"],
+			},
 			OwnedVindexQuery: "dummy_subquery",
 			KsidVindex:       ks.Vindexes["hash"],
 			KsidLength:       1,
@@ -390,8 +392,10 @@ func TestUpdateEqualMultiColChangedVindex(t *testing.T) {
 				Vindex:   ks.Vindexes["rg_vdx"],
 				Values:   []evalengine.Expr{evalengine.NewLiteralInt(1), evalengine.NewLiteralInt(2)},
 			},
-			Query:            "dummy_update",
-			Table:            ks.Tables["rg_tbl"],
+			Query: "dummy_update",
+			Table: []*vindexes.Table{
+				ks.Tables["rg_tbl"],
+			},
 			OwnedVindexQuery: "dummy_subquery",
 			KsidVindex:       ks.Vindexes["rg_vdx"],
 			KsidLength:       2,
@@ -509,8 +513,10 @@ func TestUpdateScatterChangedVindex(t *testing.T) {
 				Opcode:   Scatter,
 				Keyspace: ks.Keyspace,
 			},
-			Query:            "dummy_update",
-			Table:            ks.Tables["t1"],
+			Query: "dummy_update",
+			Table: []*vindexes.Table{
+				ks.Tables["t1"],
+			},
 			OwnedVindexQuery: "dummy_subquery",
 			KsidVindex:       ks.Vindexes["hash"],
 			KsidLength:       1,
@@ -703,8 +709,10 @@ func TestUpdateInChangedVindex(t *testing.T) {
 					evalengine.NewLiteralInt(2),
 				}},
 			},
-			Query:            "dummy_update",
-			Table:            ks.Tables["t1"],
+			Query: "dummy_update",
+			Table: []*vindexes.Table{
+				ks.Tables["t1"],
+			},
 			OwnedVindexQuery: "dummy_subquery",
 			KsidVindex:       ks.Vindexes["hash"],
 			KsidLength:       1,
@@ -832,8 +840,10 @@ func TestUpdateInChangedVindexMultiCol(t *testing.T) {
 					evalengine.NewLiteralInt(3),
 				},
 			},
-			Query:            "dummy_update",
-			Table:            ks.Tables["rg_tbl"],
+			Query: "dummy_update",
+			Table: []*vindexes.Table{
+				ks.Tables["rg_tbl"],
+			},
 			OwnedVindexQuery: "dummy_subquery",
 			KsidVindex:       ks.Vindexes["rg_vdx"],
 			KsidLength:       2,
diff --git a/go/vt/vtgate/evalengine/comparisons_test.go b/go/vt/vtgate/evalengine/comparisons_test.go
index 27be0cff2c..68daa64024 100644
--- a/go/vt/vtgate/evalengine/comparisons_test.go
+++ b/go/vt/vtgate/evalengine/comparisons_test.go
@@ -26,6 +26,7 @@ import (
 
 	"vitess.io/vitess/go/mysql/collations"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vterrors"
 
@@ -45,13 +46,21 @@ type testCase struct {
 }
 
 var (
-	T = true
-	F = false
+	T            = true
+	F            = false
+	collationEnv *collations.Environment
 )
 
+func init() {
+	// We require MySQL 8.0 collations for the comparisons in the tests
+	mySQLVersion := "8.0.0"
+	servenv.MySQLServerVersion = &mySQLVersion
+	collationEnv = collations.NewEnvironment(mySQLVersion)
+}
+
 func defaultCollation() collations.TypedCollation {
 	return collations.TypedCollation{
-		Collation:    collations.Local().LookupByName("utf8mb4_bin").ID(),
+		Collation:    collationEnv.LookupByName("utf8mb4_bin").ID(),
 		Coercibility: collations.CoerceImplicit,
 		Repertoire:   collations.RepertoireASCII,
 	}
@@ -1053,7 +1062,7 @@ func TestNullComparisons(t *testing.T) {
 }
 
 func TestNullsafeCompare(t *testing.T) {
-	collation := collations.Local().LookupByName("utf8mb4_general_ci").ID()
+	collation := collationEnv.LookupByName("utf8mb4_general_ci").ID()
 	tcases := []struct {
 		v1, v2 sqltypes.Value
 		out    int
@@ -1155,7 +1164,7 @@ func TestNullsafeCompare(t *testing.T) {
 }
 
 func getCollationID(collation string) collations.ID {
-	id, _ := collations.Local().LookupID(collation)
+	id, _ := collationEnv.LookupID(collation)
 	return id
 }
 
diff --git a/go/vt/vtgate/evalengine/integration/comparison_test.go b/go/vt/vtgate/evalengine/integration/comparison_test.go
index f55d5881ae..cecf7dc4a4 100644
--- a/go/vt/vtgate/evalengine/integration/comparison_test.go
+++ b/go/vt/vtgate/evalengine/integration/comparison_test.go
@@ -28,9 +28,19 @@ import (
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/sqltypes"
+	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
 )
 
+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
+	collationEnv = collations.NewEnvironment(mySQLVersion)
+}
+
 func perm(a []string, f func([]string)) {
 	perm1(a, f, 0)
 }
@@ -126,7 +136,7 @@ func compareRemoteExpr(t *testing.T, conn *mysql.Conn, expr string) {
 				// TODO: passthrough proper collations for nullable fields
 				remoteCollation = collations.CollationBinaryID
 			} else {
-				remoteCollation = collations.Local().LookupByName(remote.Rows[0][1].ToString()).ID()
+				remoteCollation = collationEnv.LookupByName(remote.Rows[0][1].ToString()).ID()
 			}
 		}
 	}
diff --git a/go/vt/vtgate/planbuilder/testdata/dml_cases.txt b/go/vt/vtgate/planbuilder/testdata/dml_cases.txt
index d01635938c..8357e9a678 100644
--- a/go/vt/vtgate/planbuilder/testdata/dml_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/dml_cases.txt
@@ -22,7 +22,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "update m1 set val = 1"
+    "Query": "update m1 set val = 1",
+    "Table": "m1"
   }
 }
 Gen4 plan same as above
@@ -41,7 +42,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "update unsharded set val = 1"
+    "Query": "update unsharded set val = 1",
+    "Table": "unsharded"
   }
 }
 Gen4 plan same as above
@@ -60,7 +62,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "update unsharded set col = (select col from unsharded limit 1)"
+    "Query": "update unsharded set col = (select col from unsharded limit 1)",
+    "Table": "unsharded"
   }
 }
 Gen4 plan same as above
@@ -79,7 +82,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "update unsharded set col = (select id from unsharded union select id from unsharded)"
+    "Query": "update unsharded set col = (select id from unsharded union select id from unsharded)",
+    "Table": "unsharded"
   }
 }
 Gen4 plan same as above
@@ -98,7 +102,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "update unsharded set col = (select id from unsharded as a join unsharded as b on a.id = b.id)"
+    "Query": "update unsharded set col = (select id from unsharded as a join unsharded as b on a.id = b.id)",
+    "Table": "unsharded"
   }
 }
 Gen4 plan same as above
@@ -117,7 +122,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "update unsharded as foo left join (select id from unsharded where col is not null order by col desc limit 10) as keepers on foo.id = keepers.id set col1 = 'asdf' where keepers.id is null and foo.col is not null and foo.col \u003c 1000"
+    "Query": "update unsharded as foo left join (select id from unsharded where col is not null order by col desc limit 10) as keepers on foo.id = keepers.id set col1 = 'asdf' where keepers.id is null and foo.col is not null and foo.col \u003c 1000",
+    "Table": "unsharded"
   }
 }
 Gen4 plan same as above
@@ -179,7 +185,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "update unsharded_a set a = (select a from unsharded as route2)"
+    "Query": "update unsharded_a set a = (select a from unsharded as route2)",
+    "Table": "unsharded, unsharded_a"
   }
 }
 Gen4 plan same as above
@@ -198,7 +205,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "delete from unsharded"
+    "Query": "delete from unsharded",
+    "Table": "unsharded"
   }
 }
 Gen4 plan same as above
@@ -729,7 +737,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "delete a from unsharded_a as a, unsharded_b as b where a.id = b.id and b.val = 1"
+    "Query": "delete a from unsharded_a as a, unsharded_b as b where a.id = b.id and b.val = 1",
+    "Table": "unsharded_a, unsharded_b"
   }
 }
 Gen4 plan same as above
@@ -748,7 +757,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "delete a from unsharded_a as a join unsharded_b as b on a.id = b.id where b.val = 1"
+    "Query": "delete a from unsharded_a as a join unsharded_b as b on a.id = b.id where b.val = 1",
+    "Table": "unsharded_a, unsharded_b"
   }
 }
 Gen4 plan same as above
@@ -767,7 +777,28 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "delete foo from unsharded as foo left join (select id from unsharded where col is not null order by col desc limit 10) as keepers on foo.id = keepers.id where keepers.id is null and foo.col is not null and foo.col \u003c 1000"
+    "Query": "delete foo from unsharded as foo left join (select id from unsharded where col is not null order by col desc limit 10) as keepers on foo.id = keepers.id where keepers.id is null and foo.col is not null and foo.col \u003c 1000",
+    "Table": "unsharded"
+  }
+}
+Gen4 plan same as above
+
+#delete with join from multi table join subquery
+"delete foo from unsharded as foo left join (select id from unsharded a join unsharded_b b on a.user_id = b.user_id) as keepers on foo.id = keepers.id where keepers.id is null and foo.col is not null and foo.col < 1000"
+{
+  "QueryType": "DELETE",
+  "Original": "delete foo from unsharded as foo left join (select id from unsharded a join unsharded_b b on a.user_id = b.user_id) as keepers on foo.id = keepers.id where keepers.id is null and foo.col is not null and foo.col \u003c 1000",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "Unsharded",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "TargetTabletType": "PRIMARY",
+    "MultiShardAutocommit": false,
+    "Query": "delete foo from unsharded as foo left join (select id from unsharded as a join unsharded_b as b on a.user_id = b.user_id) as keepers on foo.id = keepers.id where keepers.id is null and foo.col is not null and foo.col \u003c 1000",
+    "Table": "unsharded, unsharded, unsharded_b"
   }
 }
 Gen4 plan same as above
@@ -813,7 +844,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "delete from unsharded_a where a = (select a from unsharded as route2)"
+    "Query": "delete from unsharded_a where a = (select a from unsharded as route2)",
+    "Table": "unsharded, unsharded_a"
   }
 }
 Gen4 plan same as above
@@ -875,7 +907,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "update unsharded_a as a join unsharded_b as b on a.id = b.id set a.val = 'foo' where b.val = 1"
+    "Query": "update unsharded_a as a join unsharded_b as b on a.id = b.id set a.val = 'foo' where b.val = 1",
+    "Table": "unsharded_a, unsharded_b"
   }
 }
 Gen4 plan same as above
@@ -894,7 +927,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "update unsharded_a as a, unsharded_b as b set a.val = 'foo' where a.id = b.id and b.val = 1"
+    "Query": "update unsharded_a as a, unsharded_b as b set a.val = 'foo' where a.id = b.id and b.val = 1",
+    "Table": "unsharded_a, unsharded_b"
   }
 }
 Gen4 plan same as above
@@ -2431,7 +2465,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "update unsharded set col = (select id from unsharded_a where id = unsharded.col) where col = (select id from unsharded_b)"
+    "Query": "update unsharded set col = (select id from unsharded_a where id = unsharded.col) where col = (select id from unsharded_b)",
+    "Table": "unsharded, unsharded_a, unsharded_b"
   }
 }
 Gen4 plan same as above
@@ -2450,7 +2485,8 @@ Gen4 plan same as above
     },
     "TargetTabletType": "PRIMARY",
     "MultiShardAutocommit": false,
-    "Query": "delete from unsharded where col = (select id from unsharded_a where id = unsharded.col)"
+    "Query": "delete from unsharded where col = (select id from unsharded_a where id = unsharded.col)",
+    "Table": "unsharded, unsharded_a"
   }
 }
 Gen4 plan same as above
diff --git a/go/vt/vtgate/planbuilder/testdata/select_cases.txt b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
index 20d531e9f0..b1093204ef 100644
--- a/go/vt/vtgate/planbuilder/testdata/select_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
@@ -3530,3 +3530,105 @@ Gen4 plan same as above
 }
 Gen4 plan same as above
 
+# Predicate in apply join which is merged
+"select user.col, user_metadata.user_id from user join user_extra on user.col = user_extra.col join user_metadata on user_extra.user_id = user_metadata.user_id where user.textcol1 = 'alice@gmail.com'"
+{
+  "QueryType": "SELECT",
+  "Original": "select user.col, user_metadata.user_id from user join user_extra on user.col = user_extra.col join user_metadata on user_extra.user_id = user_metadata.user_id where user.textcol1 = 'alice@gmail.com'",
+  "Instructions": {
+    "OperatorType": "Join",
+    "Variant": "Join",
+    "JoinColumnIndexes": "L:0,R:0",
+    "JoinVars": {
+      "user_extra_user_id": 1
+    },
+    "TableName": "`user`_user_extra_user_metadata",
+    "Inputs": [
+      {
+        "OperatorType": "Join",
+        "Variant": "Join",
+        "JoinColumnIndexes": "L:0,R:0",
+        "JoinVars": {
+          "user_col": 0
+        },
+        "TableName": "`user`_user_extra",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "Scatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select `user`.col from `user` where 1 != 1",
+            "Query": "select `user`.col from `user` where `user`.textcol1 = 'alice@gmail.com'",
+            "Table": "`user`"
+          },
+          {
+            "OperatorType": "Route",
+            "Variant": "Scatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select user_extra.user_id from user_extra where 1 != 1",
+            "Query": "select user_extra.user_id from user_extra where user_extra.col = :user_col",
+            "Table": "user_extra"
+          }
+        ]
+      },
+      {
+        "OperatorType": "Route",
+        "Variant": "EqualUnique",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select user_metadata.user_id from user_metadata where 1 != 1",
+        "Query": "select user_metadata.user_id from user_metadata where user_metadata.user_id = :user_extra_user_id",
+        "Table": "user_metadata",
+        "Values": [
+          ":user_extra_user_id"
+        ],
+        "Vindex": "user_index"
+      }
+    ]
+  }
+}
+{
+  "QueryType": "SELECT",
+  "Original": "select user.col, user_metadata.user_id from user join user_extra on user.col = user_extra.col join user_metadata on user_extra.user_id = user_metadata.user_id where user.textcol1 = 'alice@gmail.com'",
+  "Instructions": {
+    "OperatorType": "Join",
+    "Variant": "Join",
+    "JoinColumnIndexes": "L:0,R:0",
+    "JoinVars": {
+      "user_col": 0
+    },
+    "TableName": "`user`_user_extra, user_metadata",
+    "Inputs": [
+      {
+        "OperatorType": "Route",
+        "Variant": "Scatter",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select `user`.col from `user` where 1 != 1",
+        "Query": "select `user`.col from `user` where `user`.textcol1 = 'alice@gmail.com'",
+        "Table": "`user`"
+      },
+      {
+        "OperatorType": "Route",
+        "Variant": "Scatter",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select user_metadata.user_id from user_extra, user_metadata where 1 != 1",
+        "Query": "select user_metadata.user_id from user_extra, user_metadata where user_extra.col = :user_col and user_extra.user_id = user_metadata.user_id",
+        "Table": "user_extra, user_metadata"
+      }
+    ]
+  }
+}
diff --git a/go/vt/vtgate/semantics/analyzer_test.go b/go/vt/vtgate/semantics/analyzer_test.go
index 8f9d4f24d4..cfb7d0d229 100644
--- a/go/vt/vtgate/semantics/analyzer_test.go
+++ b/go/vt/vtgate/semantics/analyzer_test.go
@@ -1361,26 +1361,36 @@ func TestSingleUnshardedKeyspace(t *testing.T) {
 	tests := []struct {
 		query     string
 		unsharded *vindexes.Keyspace
+		tables    []*vindexes.Table
 	}{
 		{
 			query:     "select 1 from t, t1",
 			unsharded: nil, // both tables are unsharded, but from different keyspaces
+			tables:    nil,
 		}, {
 			query:     "select 1 from t2",
 			unsharded: nil,
+			tables:    nil,
 		}, {
 			query:     "select 1 from t, t2",
 			unsharded: nil,
+			tables:    nil,
 		}, {
 			query:     "select 1 from t as A, t as B",
 			unsharded: ks1,
+			tables: []*vindexes.Table{
+				{Keyspace: ks1, Name: sqlparser.NewIdentifierCS("t")},
+				{Keyspace: ks1, Name: sqlparser.NewIdentifierCS("t")},
+			},
 		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.query, func(t *testing.T) {
 			_, semTable := parseAndAnalyze(t, test.query, "d")
-			assert.Equal(t, test.unsharded, semTable.SingleUnshardedKeyspace())
+			queryIsUnsharded, tables := semTable.SingleUnshardedKeyspace()
+			assert.Equal(t, test.unsharded, queryIsUnsharded)
+			assert.Equal(t, test.tables, tables)
 		})
 	}
 }
diff --git a/go/vt/vttablet/tabletmanager/tm_init_test.go b/go/vt/vttablet/tabletmanager/tm_init_test.go
index 90a138977c..cccdcb91d7 100644
--- a/go/vt/vttablet/tabletmanager/tm_init_test.go
+++ b/go/vt/vttablet/tabletmanager/tm_init_test.go
@@ -25,6 +25,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"vitess.io/vitess/go/mysql"
+	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/mysql/fakesqldb"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/sync2"
@@ -42,6 +43,12 @@ import (
 	vschemapb "vitess.io/vitess/go/vt/proto/vschema"
 )
 
+var (
+	dbServerVersion = "5.7.0"
+	charsetName     = "utf8mb4"
+	dbsvCollID      = collations.NewEnvironment(dbServerVersion).DefaultCollationForCharset(charsetName).ID()
+)
+
 func TestStartBuildTabletFromInput(t *testing.T) {
 	alias := &topodatapb.TabletAlias{
 		Cell: "cell",
@@ -49,7 +56,6 @@ func TestStartBuildTabletFromInput(t *testing.T) {
 	}
 	port := int32(12)
 	grpcport := int32(34)
-	dbServerVersion := "5.7.0"
 
 	// Hostname should be used as is.
 	*tabletHostname = "foo"
@@ -71,7 +77,7 @@ func TestStartBuildTabletFromInput(t *testing.T) {
 		Tags:                 map[string]string{},
 		DbNameOverride:       "aa",
 		DbServerVersion:      dbServerVersion,
-		DefaultConnCollation: 255,
+		DefaultConnCollation: uint32(dbsvCollID),
 	}
 
 	gotTablet, err := BuildTabletFromInput(alias, port, grpcport, dbServerVersion, nil)
@@ -131,7 +137,6 @@ func TestBuildTabletFromInputWithBuildTags(t *testing.T) {
 	}
 	port := int32(12)
 	grpcport := int32(34)
-	dbServerVersion := "5.7.0"
 
 	// Hostname should be used as is.
 	*tabletHostname = "foo"
@@ -154,8 +159,8 @@ func TestBuildTabletFromInputWithBuildTags(t *testing.T) {
 		Type:                 topodatapb.TabletType_REPLICA,
 		Tags:                 servenv.AppVersion.ToStringMap(),
 		DbNameOverride:       "aa",
-		DbServerVersion:      "5.7.0",
-		DefaultConnCollation: 255,
+		DbServerVersion:      dbServerVersion,
+		DefaultConnCollation: uint32(dbsvCollID),
 	}
 
 	gotTablet, err := BuildTabletFromInput(alias, port, grpcport, dbServerVersion, nil)