Repository: vitessio/vitess
Pull request #10727 (2022-07-18 11:39:17)

ApplySchema: renew keyspace lock while iterating SQLs
## Description

Followup to #10719. Before applying DDLs, `TabletExecutor` acquires a _keyspace_ lock. It then iterates all SQLs, applies them, and finally releases the lock.

When there's a large number of SQLs, the iteration can take a lon gtime. #10719 already reduces that time. However, applying SQLs can still take longer than the lock timeout. What we've seen happening was that SQLs were applied, but when it came to unlocking the keyspace, the lock was expired and the unlock operation failed.

In this PR we renew the lock on each SQL iteration. This is a naive implementation and perhaps too aggressive. We could later refine this to renew once per X seconds, for example. **UPDATE:** now calls are rate limited.

With this change, the lock will not expire on reasonable DDLs. Specifically, Online DDLs are very quick and are never expected to run long. Large `direct` `ALTER`s may still exceed lock timeout. This PR does not attempt to solve that scenario.

We have not changes the tests here. We've changes internal implementation of `CheckKeyspaceLocked()` and a re expecting to see all tests passing. In case of a problem, we will restore original behavior of this function, and create a new function which we will then use in `TabletExecutor`. **U{DATE**: refactored into a new function.

## Related Issue(s)

#6926 
#10719 

## Checklist

-   [ ] "Backport me!" label has been added if this change should be backported
-   [ ] Tests were added or are not required
-   [ ] Documentation was added or is not required

cc @dbussink 

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.

shlomi-noach: Refactored: restored original `CheckKeyspaceLocked()` function, created new `CheckKeyspaceLockedAndRenew()`. We now call it rate-limited so as not to overwhelm `ApplySchema` and `topo`.

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/timer/rate_limiter.go b/go/timer/rate_limiter.go
index 20304a03b3..d9369f9aef 100644
--- a/go/timer/rate_limiter.go
+++ b/go/timer/rate_limiter.go
@@ -46,6 +46,7 @@ func NewRateLimiter(d time.Duration) *RateLimiter {
 			select {
 			case <-ctx.Done():
 				ticker.Stop()
+				return
 			case <-ticker.C:
 				atomic.StoreInt64(&r.tickerValue, r.tickerValue+1)
 			}
diff --git a/go/vt/schemamanager/tablet_executor.go b/go/vt/schemamanager/tablet_executor.go
index 7cf3945286..aee9241d21 100644
--- a/go/vt/schemamanager/tablet_executor.go
+++ b/go/vt/schemamanager/tablet_executor.go
@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"vitess.io/vitess/go/sync2"
+	"vitess.io/vitess/go/timer"
 	"vitess.io/vitess/go/vt/logutil"
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	tabletmanagerdatapb "vitess.io/vitess/go/vt/proto/tabletmanagerdata"
@@ -31,6 +32,7 @@ import (
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/topo"
 	"vitess.io/vitess/go/vt/vtctl/schematools"
+	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vttablet/tmclient"
 )
 
@@ -332,7 +334,7 @@ func (exec *TabletExecutor) Execute(ctx context.Context, sqls []string) *Execute
 	// keyspace-wide operations like resharding migrations.
 	ctx, unlock, lockErr := exec.ts.LockKeyspace(ctx, exec.keyspace, "ApplySchemaKeyspace")
 	if lockErr != nil {
-		execResult.ExecutorErr = lockErr.Error()
+		execResult.ExecutorErr = vterrors.Wrapf(lockErr, "lockErr in ApplySchemaKeyspace %v", exec.keyspace).Error()
 		return &execResult
 	}
 	defer func() {
@@ -341,7 +343,7 @@ func (exec *TabletExecutor) Execute(ctx context.Context, sqls []string) *Execute
 		var unlockErr error
 		unlock(&unlockErr)
 		if execResult.ExecutorErr == "" && unlockErr != nil {
-			execResult.ExecutorErr = unlockErr.Error()
+			execResult.ExecutorErr = vterrors.Wrapf(unlockErr, "unlockErr in ApplySchemaKeyspace %v", exec.keyspace).Error()
 		}
 	}()
 
@@ -357,8 +359,16 @@ func (exec *TabletExecutor) Execute(ctx context.Context, sqls []string) *Execute
 	}
 	providedUUID := ""
 
+	rl := timer.NewRateLimiter(*topo.RemoteOperationTimeout / 4)
+	defer rl.Stop()
+
 	syncOperationExecuted := false
 	for index, sql := range sqls {
+		// Attempt to renew lease:
+		if err := rl.Do(func() error { return topo.CheckKeyspaceLockedAndRenew(ctx, exec.keyspace) }); err != nil {
+			execResult.ExecutorErr = vterrors.Wrapf(err, "CheckKeyspaceLocked in ApplySchemaKeyspace %v", exec.keyspace).Error()
+			return &execResult
+		}
 		execResult.CurSQLIndex = index
 		if exec.hasProvidedUUIDs() {
 			providedUUID = exec.uuids[index]
@@ -404,7 +414,7 @@ func (exec *TabletExecutor) Execute(ctx context.Context, sqls []string) *Execute
 					result.Shard,
 					result.Position,
 					concurrency,
-					false, /* includePrimary */
+					true, /* includePrimary */
 				)
 			}(result)
 		}
diff --git a/go/vt/topo/locks.go b/go/vt/topo/locks.go
index 9c6d6dbd70..ad43a87319 100644
--- a/go/vt/topo/locks.go
+++ b/go/vt/topo/locks.go
@@ -208,6 +208,26 @@ func CheckKeyspaceLocked(ctx context.Context, keyspace string) error {
 	return nil
 }
 
+// CheckKeyspaceLockedAndRenew can be called on a context to make sure we have the lock
+// for a given keyspace. The function also attempts to renew the lock.
+func CheckKeyspaceLockedAndRenew(ctx context.Context, keyspace string) error {
+	// extract the locksInfo pointer
+	i, ok := ctx.Value(locksKey).(*locksInfo)
+	if !ok {
+		return vterrors.Errorf(vtrpc.Code_INVALID_ARGUMENT, "keyspace %v is not locked (no locksInfo)", keyspace)
+	}
+	i.mu.Lock()
+	defer i.mu.Unlock()
+
+	// find the individual entry
+	entry, ok := i.info[keyspace]
+	if !ok {
+		return vterrors.Errorf(vtrpc.Code_INVALID_ARGUMENT, "keyspace %v is not locked (no lockInfo in map)", keyspace)
+	}
+	// try renewing lease:
+	return entry.lockDescriptor.Check(ctx)
+}
+
 // lockKeyspace will lock the keyspace in the topology server.
 // unlockKeyspace should be called if this returns no error.
 func (l *Lock) lockKeyspace(ctx context.Context, ts *Server, keyspace string) (LockDescriptor, error) {
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
 }


Test patch:
diff --git a/.github/workflows/unit_test_mariadb102.yml b/.github/workflows/unit_test_mariadb102.yml
deleted file mode 100644
index 43bf4935b6..0000000000
--- a/.github/workflows/unit_test_mariadb102.yml
+++ /dev/null
@@ -1,95 +0,0 @@
-# DO NOT MODIFY: THIS FILE IS GENERATED USING "make generate_ci_workflows"
-
-name: Unit Test (mariadb102)
-on: [push, pull_request]
-concurrency:
-  group: format('{0}-{1}', ${{ github.ref }}, 'Unit Test (mariadb102)')
-  cancel-in-progress: true
-
-jobs:
-  test:
-    runs-on: ubuntu-18.04
-
-    steps:
-    - name: Check out code
-      uses: actions/checkout@v2
-
-    - name: Check for changes in relevant files
-      uses: frouioui/paths-filter@main
-      id: changes
-      with:
-        token: ''
-        filters: |
-          unit_tests:
-            - 'go/**'
-            - 'test.go'
-            - 'Makefile'
-            - 'build.env'
-            - 'go.[sumod]'
-            - 'proto/*.proto'
-            - 'tools/**'
-            - 'config/**'
-            - 'bootstrap.sh'
-            - '.github/workflows/**'
-
-    - name: Set up Go
-      if: steps.changes.outputs.unit_tests == 'true'
-      uses: actions/setup-go@v2
-      with:
-        go-version: 1.18.3
-
-    - name: Tune the OS
-      if: steps.changes.outputs.unit_tests == 'true'
-      run: |
-        echo '1024 65535' | sudo tee -a /proc/sys/net/ipv4/ip_local_port_range
-        # Increase the asynchronous non-blocking I/O. More information at https://dev.mysql.com/doc/refman/5.7/en/innodb-parameters.html#sysvar_innodb_use_native_aio
-        echo "fs.aio-max-nr = 1048576" | sudo tee -a /etc/sysctl.conf
-        sudo sysctl -p /etc/sysctl.conf
-
-    - name: Get dependencies
-      if: steps.changes.outputs.unit_tests == 'true'
-      run: |
-        export DEBIAN_FRONTEND="noninteractive"
-        sudo apt-get update
-
-        # !mysql57
-
-        # Uninstall any previously installed MySQL first
-        sudo systemctl stop apparmor
-        sudo DEBIAN_FRONTEND="noninteractive" apt-get remove -y --purge mysql-server mysql-client mysql-common
-        sudo apt-get -y autoremove
-        sudo apt-get -y autoclean
-        sudo deluser mysql
-        sudo rm -rf /var/lib/mysql
-        sudo rm -rf /etc/mysql
-
-        # mariadb102
-        sudo apt-get install -y software-properties-common
-        sudo apt-key adv --recv-keys --keyserver hkp://keyserver.ubuntu.com:80 0xF1656F24C74CD1D8
-        sudo add-apt-repository 'deb [arch=amd64,arm64,ppc64el] http://nyc2.mirrors.digitalocean.com/mariadb/repo/10.2/ubuntu bionic main'
-        sudo apt update
-        sudo DEBIAN_FRONTEND="noninteractive" apt install -y mariadb-server
-
-        sudo apt-get install -y make unzip g++ curl git wget ant openjdk-8-jdk eatmydata
-        sudo service mysql stop
-        sudo bash -c "echo '/usr/sbin/mysqld { }' > /etc/apparmor.d/usr.sbin.mysqld" # https://bugs.launchpad.net/ubuntu/+source/mariadb-10.1/+bug/1806263
-        sudo ln -s /etc/apparmor.d/usr.sbin.mysqld /etc/apparmor.d/disable/
-        sudo apparmor_parser -R /etc/apparmor.d/usr.sbin.mysqld || echo "could not remove mysqld profile"
-
-        mkdir -p dist bin
-        curl -L https://github.com/coreos/etcd/releases/download/v3.3.10/etcd-v3.3.10-linux-amd64.tar.gz | tar -zxC dist
-        mv dist/etcd-v3.3.10-linux-amd64/{etcd,etcdctl} bin/
-
-        go mod download
-        go install golang.org/x/tools/cmd/goimports@latest
-
-    - name: Run make tools
-      if: steps.changes.outputs.unit_tests == 'true'
-      run: |
-        make tools
-
-    - name: Run test
-      if: steps.changes.outputs.unit_tests == 'true'
-      timeout-minutes: 30
-      run: |
-        eatmydata -- make unit_test
diff --git a/.github/workflows/upgrade_downgrade_test_backups_e2e.yml b/.github/workflows/upgrade_downgrade_test_backups_e2e.yml
index fb7c474e1e..f7bc41e50e 100644
--- a/.github/workflows/upgrade_downgrade_test_backups_e2e.yml
+++ b/.github/workflows/upgrade_downgrade_test_backups_e2e.yml
@@ -29,10 +29,10 @@ jobs:
           echo "::set-output name=previous_release_ref::${previous_release_ref}"
 
   upgrade_downgrade_test_e2e:
-    timeout-minutes: 60
     if: always() && needs.get_previous_release.result == 'success'
     name: Run Upgrade Downgrade Test
     runs-on: ubuntu-18.04
+    timeout-minutes: 60
     needs:
       - get_previous_release
 
diff --git a/.github/workflows/upgrade_downgrade_test_query_serving_queries.yml b/.github/workflows/upgrade_downgrade_test_query_serving_queries.yml
index ad225c3b0e..056211acb8 100644
--- a/.github/workflows/upgrade_downgrade_test_query_serving_queries.yml
+++ b/.github/workflows/upgrade_downgrade_test_query_serving_queries.yml
@@ -35,6 +35,7 @@ jobs:
     if: always() && (needs.get_previous_release.result == 'success')
     name: Run Upgrade Downgrade Test
     runs-on: ubuntu-latest
+    timeout-minutes: 60
     needs:
       - get_upgrade_downgrade_label
       - get_previous_release
diff --git a/.github/workflows/upgrade_downgrade_test_query_serving_schema.yml b/.github/workflows/upgrade_downgrade_test_query_serving_schema.yml
index 9ebf63ff09..6c989592ba 100644
--- a/.github/workflows/upgrade_downgrade_test_query_serving_schema.yml
+++ b/.github/workflows/upgrade_downgrade_test_query_serving_schema.yml
@@ -35,6 +35,7 @@ jobs:
     if: always() && (needs.get_previous_release.result == 'success')
     name: Run Upgrade Downgrade Test
     runs-on: ubuntu-latest
+    timeout-minutes: 60
     needs:
       - get_previous_release
 
diff --git a/.github/workflows/upgrade_downgrade_test_reparent_old_vtctl.yml b/.github/workflows/upgrade_downgrade_test_reparent_old_vtctl.yml
index 4628d2be23..ec8cdeb8c1 100644
--- a/.github/workflows/upgrade_downgrade_test_reparent_old_vtctl.yml
+++ b/.github/workflows/upgrade_downgrade_test_reparent_old_vtctl.yml
@@ -35,6 +35,7 @@ jobs:
     if: always() && (needs.get_previous_release.result == 'success')
     name: Run Upgrade Downgrade Test
     runs-on: ubuntu-latest
+    timeout-minutes: 60
     needs:
       - get_previous_release
 
diff --git a/.github/workflows/upgrade_downgrade_test_reparent_old_vttablet.yml b/.github/workflows/upgrade_downgrade_test_reparent_old_vttablet.yml
index 53a7698252..ed169ad791 100644
--- a/.github/workflows/upgrade_downgrade_test_reparent_old_vttablet.yml
+++ b/.github/workflows/upgrade_downgrade_test_reparent_old_vttablet.yml
@@ -37,6 +37,7 @@ jobs:
     if: always() && (needs.get_previous_release.result == 'success')
     name: Run Upgrade Downgrade Test
     runs-on: ubuntu-latest
+    timeout-minutes: 60
     needs:
       - get_previous_release
 
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 9bff176a9e..6f103e6051 100644
--- a/go/test/endtoend/reparent/emergencyreparent/ers_test.go
+++ b/go/test/endtoend/reparent/emergencyreparent/ers_test.go
@@ -31,7 +31,7 @@ import (
 
 func TestTrivialERS(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -56,7 +56,7 @@ func TestTrivialERS(t *testing.T) {
 
 func TestReparentIgnoreReplicas(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 	var err error
@@ -98,7 +98,7 @@ func TestReparentIgnoreReplicas(t *testing.T) {
 
 func TestReparentDownPrimary(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -134,7 +134,7 @@ func TestReparentDownPrimary(t *testing.T) {
 
 func TestReparentNoChoiceDownPrimary(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 	var err error
@@ -170,7 +170,7 @@ func TestReparentNoChoiceDownPrimary(t *testing.T) {
 func TestSemiSyncSetupCorrectly(t *testing.T) {
 	t.Run("semi-sync enabled", func(t *testing.T) {
 		defer cluster.PanicHandler(t)
-		clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+		clusterInstance := utils.SetupReparentCluster(t, true)
 		defer utils.TeardownCluster(clusterInstance)
 		tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -198,7 +198,7 @@ func TestSemiSyncSetupCorrectly(t *testing.T) {
 
 	t.Run("semi-sync disabled", func(t *testing.T) {
 		defer cluster.PanicHandler(t)
-		clusterInstance := utils.SetupReparentClusterLegacy(t, false)
+		clusterInstance := utils.SetupReparentCluster(t, false)
 		defer utils.TeardownCluster(clusterInstance)
 		tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
diff --git a/go/test/endtoend/reparent/plannedreparent/reparent_test.go b/go/test/endtoend/reparent/plannedreparent/reparent_test.go
index 2e6b67c68d..ce62a7bc11 100644
--- a/go/test/endtoend/reparent/plannedreparent/reparent_test.go
+++ b/go/test/endtoend/reparent/plannedreparent/reparent_test.go
@@ -34,7 +34,7 @@ import (
 
 func TestPrimaryToSpareStateChangeImpossible(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -46,7 +46,7 @@ func TestPrimaryToSpareStateChangeImpossible(t *testing.T) {
 
 func TestReparentCrossCell(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -60,7 +60,7 @@ func TestReparentCrossCell(t *testing.T) {
 
 func TestReparentGraceful(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -83,7 +83,7 @@ func TestReparentGraceful(t *testing.T) {
 // TestPRSWithDrainedLaggingTablet tests that PRS succeeds even if we have a lagging drained tablet
 func TestPRSWithDrainedLaggingTablet(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -110,7 +110,7 @@ func TestPRSWithDrainedLaggingTablet(t *testing.T) {
 
 func TestReparentReplicaOffline(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -127,7 +127,7 @@ func TestReparentReplicaOffline(t *testing.T) {
 
 func TestReparentAvoid(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 	utils.DeleteTablet(t, clusterInstance, tablets[2])
@@ -159,14 +159,14 @@ func TestReparentAvoid(t *testing.T) {
 
 func TestReparentFromOutside(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	reparentFromOutside(t, clusterInstance, false)
 }
 
 func TestReparentFromOutsideWithNoPrimary(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -255,7 +255,7 @@ func reparentFromOutside(t *testing.T, clusterInstance *cluster.LocalProcessClus
 
 func TestReparentWithDownReplica(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -298,7 +298,7 @@ func TestReparentWithDownReplica(t *testing.T) {
 
 func TestChangeTypeSemiSync(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -362,7 +362,7 @@ func TestChangeTypeSemiSync(t *testing.T) {
 
 func TestReparentDoesntHangIfPrimaryFails(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentClusterLegacy(t, true)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
diff --git a/go/test/endtoend/reparent/utils/utils.go b/go/test/endtoend/reparent/utils/utils.go
index bffcad957d..a716183017 100644
--- a/go/test/endtoend/reparent/utils/utils.go
+++ b/go/test/endtoend/reparent/utils/utils.go
@@ -67,11 +67,6 @@ var (
 
 //region cluster setup/teardown
 
-// SetupReparentClusterLegacy is used to setup the reparent cluster
-func SetupReparentClusterLegacy(t *testing.T, enableSemiSync bool) *cluster.LocalProcessCluster {
-	return setupClusterLegacy(context.Background(), t, ShardName, []string{cell1, cell2}, []int{3, 1}, enableSemiSync)
-}
-
 // SetupReparentCluster is used to setup the reparent cluster
 func SetupReparentCluster(t *testing.T, enableSemiSync bool) *cluster.LocalProcessCluster {
 	return setupCluster(context.Background(), t, ShardName, []string{cell1, cell2}, []int{3, 1}, enableSemiSync)
@@ -79,7 +74,7 @@ func SetupReparentCluster(t *testing.T, enableSemiSync bool) *cluster.LocalProce
 
 // SetupRangeBasedCluster sets up the range based cluster
 func SetupRangeBasedCluster(ctx context.Context, t *testing.T) *cluster.LocalProcessCluster {
-	return setupClusterLegacy(ctx, t, ShardName, []string{cell1}, []int{2}, true)
+	return setupCluster(ctx, t, ShardName, []string{cell1}, []int{2}, true)
 }
 
 // TeardownCluster is used to teardown the reparent cluster
@@ -202,126 +197,6 @@ func setupShard(ctx context.Context, t *testing.T, clusterInstance *cluster.Loca
 	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 {
-	var tablets []*cluster.Vttablet
-	clusterInstance := cluster.NewCluster(cells[0], Hostname)
-	keyspace := &cluster.Keyspace{Name: KeyspaceName}
-
-	durability := "none"
-	if enableSemiSync {
-		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--enable_semi_sync")
-		durability = "semi_sync"
-	}
-
-	// Start topo server
-	err := clusterInstance.StartTopo()
-	require.NoError(t, err, "Error starting topo")
-	err = clusterInstance.TopoProcess.ManageTopoDir("mkdir", "/vitess/"+cells[0])
-	require.NoError(t, err, "Error managing topo")
-	numCell := 1
-	for numCell < len(cells) {
-		err = clusterInstance.VtctlProcess.AddCellInfo(cells[numCell])
-		require.NoError(t, err, "Error managing topo")
-		numCell++
-	}
-
-	// Adding another cell in the same cluster
-	numCell = 0
-	for numCell < len(cells) {
-		i := 0
-		for i < numTablets[numCell] {
-			i++
-			tablet := clusterInstance.NewVttabletInstance("replica", 100*(numCell+1)+i, cells[numCell])
-			tablets = append(tablets, tablet)
-		}
-		numCell++
-	}
-
-	shard := &cluster.Shard{Name: shardName}
-	shard.Vttablets = tablets
-
-	clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs,
-		"--lock_tables_timeout", "5s",
-		"--init_populate_metadata",
-		"--track_schema_versions=true",
-		// disabling online-ddl for reparent tests. This is done to reduce flakiness.
-		// All the tests in this package reparent frequently between different tablets
-		// This means that Promoting a tablet to primary is sometimes immediately followed by a DemotePrimary call.
-		// In this case, the close method and initSchema method of the onlineDDL executor race.
-		// If the initSchema acquires the lock, then it takes about 30 seconds for it to run during which time the
-		// DemotePrimary rpc is stalled!
-		"--queryserver_enable_online_ddl=false",
-		// disabling active reparents on the tablet since we don't want the replication manager
-		// to fix replication if it is stopped. Some tests deliberately do that. Also, we don't want
-		// the replication manager to silently fix the replication in case ERS or PRS mess up. All the
-		// tests in this test suite should work irrespective of this flag. Each run of ERS, PRS should be
-		// setting up the replication correctly.
-		"--disable_active_reparents")
-
-	// Initialize Cluster
-	err = clusterInstance.SetupCluster(keyspace, []cluster.Shard{*shard})
-	require.NoError(t, err, "Cannot launch cluster")
-
-	//Start MySql
-	var mysqlCtlProcessList []*exec.Cmd
-	for _, shard := range clusterInstance.Keyspaces[0].Shards {
-		for _, tablet := range shard.Vttablets {
-			log.Infof("Starting MySql for tablet %v", tablet.Alias)
-			proc, err := tablet.MysqlctlProcess.StartProcess()
-			require.NoError(t, err, "Error starting start mysql")
-			mysqlCtlProcessList = append(mysqlCtlProcessList, proc)
-		}
-	}
-
-	// Wait for mysql processes to start
-	for _, proc := range mysqlCtlProcessList {
-		if err := proc.Wait(); err != nil {
-			clusterInstance.PrintMysqlctlLogFiles()
-			require.FailNow(t, "Error starting mysql: %s", err.Error())
-		}
-	}
-
-	if clusterInstance.VtctlMajorVersion >= 14 {
-		vtctldClientProcess := cluster.VtctldClientProcessInstance("localhost", clusterInstance.VtctldProcess.GrpcPort, clusterInstance.TmpDirectory)
-		out, err := vtctldClientProcess.ExecuteCommandWithOutput("SetKeyspaceDurabilityPolicy", KeyspaceName, fmt.Sprintf("--durability-policy=%s", durability))
-		require.NoError(t, err, out)
-	}
-
-	setupShardLegacy(ctx, t, clusterInstance, shardName, tablets)
-	return clusterInstance
-}
-
-func setupShardLegacy(ctx context.Context, t *testing.T, clusterInstance *cluster.LocalProcessCluster, shardName string, tablets []*cluster.Vttablet) {
-	for _, tablet := range tablets {
-		// create database
-		err := tablet.VttabletProcess.CreateDB(KeyspaceName)
-		require.NoError(t, err)
-		// Start the tablet
-		err = tablet.VttabletProcess.Setup()
-		require.NoError(t, err)
-	}
-
-	for _, tablet := range tablets {
-		err := tablet.VttabletProcess.WaitForTabletStatuses([]string{"SERVING", "NOT_SERVING"})
-		require.NoError(t, err)
-	}
-
-	// Force the replica to reparent assuming that all the datasets are identical.
-	err := clusterInstance.VtctlclientProcess.ExecuteCommand("InitShardPrimary", "--",
-		"--force", fmt.Sprintf("%s/%s", KeyspaceName, shardName), tablets[0].Alias)
-	require.NoError(t, err)
-
-	ValidateTopology(t, clusterInstance, true)
-
-	// create Tables
-	RunSQL(ctx, t, sqlSchema, tablets[0])
-
-	CheckPrimaryTablet(t, clusterInstance, tablets[0])
-
-	ValidateTopology(t, clusterInstance, false)
-	WaitForReplicationToStart(t, clusterInstance, KeyspaceName, shardName, len(tablets), true)
-}
-
 //endregion
 
 //region database queries
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/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/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/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/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)
diff --git a/test/ci_workflow_gen.go b/test/ci_workflow_gen.go
index eb15061d96..d92a376c4a 100644
--- a/test/ci_workflow_gen.go
+++ b/test/ci_workflow_gen.go
@@ -31,7 +31,6 @@ type mysqlVersion string
 const (
 	mysql57    mysqlVersion = "mysql57"
 	mysql80    mysqlVersion = "mysql80"
-	mariadb102 mysqlVersion = "mariadb102"
 	mariadb103 mysqlVersion = "mariadb103"
 
 	defaultMySQLVersion mysqlVersion = mysql57
@@ -46,7 +45,7 @@ var (
 )
 
 var (
-	unitTestDatabases = []mysqlVersion{mysql57, mysql80, mariadb102, mariadb103}
+	unitTestDatabases = []mysqlVersion{mysql57, mysql80, mariadb103}
 )
 
 const (
diff --git a/test/config.json b/test/config.json
index 30613293f0..ac925f1af2 100644
--- a/test/config.json
+++ b/test/config.json
@@ -111,7 +111,7 @@
 		},
 		"backup_only": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/backup/vtbackup"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/backup/vtbackup", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "19",
@@ -120,7 +120,7 @@
 		},
 		"backup_transform": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/backup/transform"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/backup/transform", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "19",
@@ -382,7 +382,7 @@
 		},
 		"emergencyreparent": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/reparent/emergencyreparent"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/reparent/emergencyreparent", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "ers_prs_newfeatures_heavy",
@@ -391,7 +391,7 @@
 		},
 		"plannedreparent": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/reparent/plannedreparent"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/reparent/plannedreparent", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "ers_prs_newfeatures_heavy",
@@ -569,7 +569,7 @@
 		},
 		"vtgate_queries_orderby": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/queries/orderby"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/queries/orderby", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "vtgate_queries",
@@ -578,7 +578,7 @@
 		},
 		"vtgate_queries_subquery": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/queries/subquery"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/queries/subquery", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "vtgate_queries",
@@ -587,7 +587,7 @@
 		},
 		"vtgate_queries_union": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/queries/union"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/queries/union", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "vtgate_queries",
@@ -614,7 +614,7 @@
 		},
 		"vtgate_schema": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/schema"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/schema", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "vtgate_schema",
@@ -641,7 +641,7 @@
 		},
 		"vtgate_schematracker_sharded": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/schematracker/sharded"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/schematracker/sharded", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "vtgate_schema_tracker",
@@ -650,7 +650,7 @@
 		},
 		"vtgate_schematracker_sharded_prs": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/schematracker/sharded_prs"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/schematracker/sharded_prs", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "vtgate_schema_tracker",
@@ -659,7 +659,7 @@
 		},
 		"vtgate_schematracker_unauthorized": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/schematracker/unauthorized"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/schematracker/unauthorized", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "vtgate_schema_tracker",
@@ -668,7 +668,7 @@
 		},
 		"vtgate_schematracker_unsharded": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/schematracker/unsharded"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/schematracker/unsharded", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "vtgate_schema_tracker",
@@ -803,7 +803,7 @@
 		},
 		"vtgate_vschema": {
 			"File": "unused.go",
-			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/vschema"],
+			"Args": ["vitess.io/vitess/go/test/endtoend/vtgate/vschema", "-timeout", "20m"],
 			"Command": [],
 			"Manual": false,
 			"Shard": "vtgate_vschema",