Repository: vitessio/vitess
Pull request #13938 (2023-09-07 21:35:54)

[release-15.0] Bump upgrade test to `go1.20.8`
## Description

This PR is a follow-up of the golang upgrade done in https://github.com/vitessio/vitess/pull/13932. We are bumping the golang version used in the upgrade tests to reflect the new golang version that will be used in `release-16.0`.



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.
- [ ] Ensure there is a link to an issue (except for internal cleanup and flaky test fixes), new features should have an RFC that documents use cases and test cases.

#### Tests
- [ ] Bug fixes should have at least one unit or end-to-end test, enhancement and new features should have a sufficient number of tests.

#### Documentation
- [ ] Apply the `release notes (needs details)` label if users need to know about this change.
- [ ] New features should be documented.
- [ ] There should be some code comments as to why things are implemented the way they are.
- [ ] There should be a comment at the top of each new or modified test to explain what the test does.

#### New flags
- [ ] Is this flag really necessary?
- [ ] Flag names must be clear and intuitive, use dashes (`-`), and have a clear help text.

#### If a workflow is added or modified:
- [ ] Each item in `Jobs` should be named in order to mark it as `required`.
- [ ] If the workflow needs to be marked as `required`, the maintainer team must be notified.

#### Backward compatibility
- [ ] Protobuf changes should be wire-compatible.
- [ ] Changes to `_vt` tables and RPCs need to be backward compatible.
- [ ] RPC changes should be compatible with vitess-operator
- [ ] If a flag is removed, then it should also be removed from [vitess-operator](https://github.com/planetscale/vitess-operator) and [arewefastyet](https://github.com/vitessio/arewefastyet), if used there.
- [ ] `vtctl` command output order should be stable and `awk`-able.


Patch:
diff --git a/.github/workflows/code_freeze.yml b/.github/workflows/code_freeze.yml
index 381793fa84..b6dde0a022 100644
--- a/.github/workflows/code_freeze.yml
+++ b/.github/workflows/code_freeze.yml
@@ -9,4 +9,4 @@ jobs:
     steps:
     - name: Fail if Code Freeze is enabled
       run: |
-        exit 0
+        exit 1
diff --git a/.github/workflows/static_checks_etc.yml b/.github/workflows/static_checks_etc.yml
index 9bb0429644..2a7033c88c 100644
--- a/.github/workflows/static_checks_etc.yml
+++ b/.github/workflows/static_checks_etc.yml
@@ -24,12 +24,6 @@ jobs:
         if: steps.skip-workflow.outputs.skip-workflow == 'false'
         uses: actions/checkout@v3
 
-      - name: Run FOSSA scan and upload build data
-        if: steps.skip-workflow.outputs.skip-workflow == 'false'
-        uses: fossa-contrib/fossa-action@v1
-        with:
-          fossa-api-key: 76d7483ea206d530d9452e44bffe7ba8
-
       - name: Check for changes in Go files
         if: steps.skip-workflow.outputs.skip-workflow == 'false'
         uses: frouioui/paths-filter@main
@@ -93,6 +87,7 @@ jobs:
             release_notes:
               - 'changelog/**'
               - './go/tools/releases/**'
+              - '.github/workflows/static_checks_etc.yml'
 
       - name: Set up Go
         if: steps.skip-workflow.outputs.skip-workflow == 'false' && (steps.changes.outputs.go_files == 'true' || steps.changes.outputs.parser_changes == 'true' || steps.changes.outputs.proto_changes == 'true')
@@ -208,4 +203,5 @@ jobs:
           echo 'We wish to maintain a consistent changelog directory, please run `go run ./go/tools/releases/releases.go`, commit and push again.'
           echo 'Running `go run ./go/tools/releases/releases.go` on CI yields the following changes:'
           echo "$output"
-          echo ""
\ No newline at end of file
+          echo ""
+          exit 1
diff --git a/docker/k8s/Dockerfile b/docker/k8s/Dockerfile
index 30ff33952b..1967aea538 100644
--- a/docker/k8s/Dockerfile
+++ b/docker/k8s/Dockerfile
@@ -51,6 +51,7 @@ COPY --from=base /vt/bin/vtgate /vt/bin/
 COPY --from=base /vt/bin/vttablet /vt/bin/
 COPY --from=base /vt/bin/vtbackup /vt/bin/
 COPY --from=base /vt/bin/vtadmin /vt/bin/
+COPY --from=base /vt/bin/vtorc /vt/bin/
 
 # copy web admin files
 COPY --from=base $VTROOT/web /vt/web/
diff --git a/docker/k8s/vtorc/Dockerfile b/docker/k8s/vtorc/Dockerfile
new file mode 100644
index 0000000000..b62b30ee67
--- /dev/null
+++ b/docker/k8s/vtorc/Dockerfile
@@ -0,0 +1,38 @@
+# Copyright 2019 The Vitess Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+ARG VT_BASE_VER=latest
+ARG DEBIAN_VER=stable-slim
+
+FROM vitess/k8s:${VT_BASE_VER} AS k8s
+
+FROM debian:${DEBIAN_VER}
+
+# Set up Vitess environment (just enough to run pre-built Go binaries)
+ENV VTROOT /vt
+
+# Prepare directory structure.
+RUN mkdir -p /vt/bin && mkdir -p /vtdataroot
+
+# Copy certs to allow https calls
+COPY --from=k8s /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
+
+# Copy binaries
+COPY --from=k8s /vt/bin/vtorc /vt/bin/
+
+# add vitess user/group and add permissions
+RUN groupadd -r --gid 2000 vitess && \
+   useradd -r -g vitess --uid 1000 vitess && \
+   chown -R vitess:vitess /vt && \
+   chown -R vitess:vitess /vtdataroot
diff --git a/docker/release.sh b/docker/release.sh
index 80d2eabcb0..26130f15e7 100755
--- a/docker/release.sh
+++ b/docker/release.sh
@@ -21,6 +21,11 @@ do
   docker push vitess/vtadmin:$vt_base_version-$debian_version
   if [[ $debian_version == $default_debian_version ]]; then docker push vitess/vtadmin:$vt_base_version; fi
 
+  docker build --platform linux/amd64 --build-arg VT_BASE_VER=$vt_base_version --build-arg DEBIAN_VER=$debian_version-slim -t vitess/vtorc:$vt_base_version-$debian_version k8s/vtorc
+  docker tag vitess/vtorc:$vt_base_version-$debian_version vitess/vtorc:$vt_base_version
+  docker push vitess/vtorc:$vt_base_version-$debian_version
+  if [[ $debian_version == $default_debian_version ]]; then docker push vitess/vtorc:$vt_base_version; fi
+
   docker build --platform linux/amd64 --build-arg VT_BASE_VER=$vt_base_version --build-arg DEBIAN_VER=$debian_version-slim -t vitess/vtgate:$vt_base_version-$debian_version k8s/vtgate
   docker tag vitess/vtgate:$vt_base_version-$debian_version vitess/vtgate:$vt_base_version
   docker push vitess/vtgate:$vt_base_version-$debian_version
diff --git a/go/cmd/internal/docgen/docgen.go b/go/cmd/internal/docgen/docgen.go
index 6fe461e5af..16b237ea15 100644
--- a/go/cmd/internal/docgen/docgen.go
+++ b/go/cmd/internal/docgen/docgen.go
@@ -46,8 +46,10 @@ import (
 	"fmt"
 	"io/fs"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"strings"
+	"sync"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/cobra/doc"
@@ -57,6 +59,10 @@ import (
 // written to `dir`. The root command is also renamed to _index.md to remain
 // compatible with the vitessio/website content structure expectations.
 func GenerateMarkdownTree(cmd *cobra.Command, dir string) error {
+	sha, err := getCommitID("HEAD")
+	if err != nil {
+		return fmt.Errorf("failed to get commit id for HEAD: %w", err)
+	}
 	switch fi, err := os.Stat(dir); {
 	case errors.Is(err, fs.ErrNotExist):
 		if err := os.MkdirAll(dir, 0755); err != nil {
@@ -69,7 +75,7 @@ func GenerateMarkdownTree(cmd *cobra.Command, dir string) error {
 	}
 
 	recursivelyDisableAutoGenTags(cmd)
-	if err := doc.GenMarkdownTreeCustom(cmd, dir, frontmatterFilePrepender, linkHandler); err != nil {
+	if err := doc.GenMarkdownTreeCustom(cmd, dir, frontmatterFilePrepender(sha), linkHandler); err != nil {
 		return err
 	}
 
@@ -79,6 +85,120 @@ func GenerateMarkdownTree(cmd *cobra.Command, dir string) error {
 		return fmt.Errorf("failed to index doc (generated at %s) into proper position (%s): %w", rootDocPath, indexDocPath, err)
 	}
 
+	if err := anonymizeHomedir(indexDocPath); err != nil {
+		return fmt.Errorf("failed to anonymize homedir in help text for command %s: %w", indexDocPath, err)
+	}
+
+	if err := restructure(dir, dir, cmd.Name(), cmd.Commands()); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+/*
+_index.md (aka vtctldclient.md)
+vtctldclient_AddCellInfo.md
+vtctldclient_movetables.md
+vtctldclient_movetables_show.md
+
+becomes
+
+_index.md
+vtctldclient_AddCellInfo.md
+vtctldclient_movetables/
+	_index.md
+	vtctldclient_movetables_show.md
+*/
+
+func restructure(rootDir string, dir string, name string, commands []*cobra.Command) error {
+	for _, cmd := range commands {
+		fullCmdFilename := strings.Join([]string{name, cmd.Name()}, "_")
+
+		children := cmd.Commands()
+
+		switch {
+		case len(children) > 0:
+			// Command (top-level or not) with children.
+			// 1. Set up a directory for its children.
+			// 2. Move its doc into that dir as "_index.md"
+			// 3. Restructure its children.
+			cmdDir := filepath.Join(dir, fullCmdFilename)
+			if err := os.MkdirAll(cmdDir, 0755); err != nil {
+				return fmt.Errorf("failed to create subdir for %s: %w", fullCmdFilename, err)
+			}
+
+			indexFile := filepath.Join(cmdDir, "_index.md")
+			if err := os.Rename(filepath.Join(rootDir, fullCmdFilename+".md"), indexFile); err != nil {
+				return fmt.Errorf("failed to move index doc for command %s with children: %w", fullCmdFilename, err)
+			}
+
+			if err := anonymizeHomedir(indexFile); err != nil {
+				return fmt.Errorf("failed to anonymize homedir in help text for command %s: %w", indexFile, err)
+			}
+
+			if err := restructure(rootDir, cmdDir, fullCmdFilename, children); err != nil {
+				return fmt.Errorf("failed to restructure child commands for %s: %w", fullCmdFilename, err)
+			}
+		case rootDir != dir:
+			// Sub-command without children.
+			// 1. Move its doc into the directory for its parent, name unchanged.
+			if cmd.Name() == "help" {
+				// all commands with children have their own "help" subcommand,
+				// which we do not generate docs for
+				continue
+			}
+
+			oldName := filepath.Join(rootDir, fullCmdFilename+".md")
+			newName := filepath.Join(dir, fullCmdFilename+".md")
+
+			if err := os.Rename(oldName, newName); err != nil {
+				return fmt.Errorf("failed to move child command %s to its parent's dir: %w", fullCmdFilename, err)
+			}
+
+			sed := newParentLinkSedCommand(name, newName)
+			if out, err := sed.CombinedOutput(); err != nil {
+				return fmt.Errorf("failed to rewrite links to parent command in child %s: %w (extra: %s)", newName, err, out)
+			}
+
+			if err := anonymizeHomedir(newName); err != nil {
+				return fmt.Errorf("failed to anonymize homedir in help text for command %s: %w", newName, err)
+			}
+		default:
+			// Top-level command without children. Nothing to restructure.
+			continue
+		}
+	}
+
+	return nil
+}
+
+func newParentLinkSedCommand(parent string, file string) *exec.Cmd {
+	return exec.Command("sed", "-i", "", "-e", fmt.Sprintf("s:(./%s/):(../):i", parent), file)
+}
+
+var (
+	wd   string
+	once sync.Once
+)
+
+func anonymizeHomedir(file string) (err error) {
+	once.Do(func() {
+		// Only do this once per run.
+		wd, err = os.Getwd()
+	})
+	if err != nil {
+		return err
+	}
+
+	// We're replacing the stuff inside the square brackets in the example sed
+	// below:
+	// 	's:Paths to search for config files in. (default \[.*\])$:Paths to search for config files in. (default \[$WORKDIR\]):'
+	sed := exec.Command("sed", "-i", "", "-e", fmt.Sprintf("s:%s:$WORKDIR:i", wd), file)
+	if out, err := sed.CombinedOutput(); err != nil {
+		return fmt.Errorf("%w: %s", err, out)
+	}
+
 	return nil
 }
 
@@ -91,22 +211,37 @@ func recursivelyDisableAutoGenTags(root *cobra.Command) {
 	}
 }
 
+func getCommitID(ref string) (string, error) {
+	gitShow := exec.Command("git", "show", "--pretty=format:%H", "--no-patch", ref)
+	out, err := gitShow.Output()
+	if err != nil {
+		return "", err
+	}
+
+	return string(out), nil
+}
+
 const frontmatter = `---
 title: %s
 series: %s
+commit: %s
 ---
 `
 
-func frontmatterFilePrepender(filename string) string {
-	name := filepath.Base(filename)
-	base := strings.TrimSuffix(name, filepath.Ext(name))
+func frontmatterFilePrepender(sha string) func(filename string) string {
+	return func(filename string) string {
+		name := filepath.Base(filename)
+		base := strings.TrimSuffix(name, filepath.Ext(name))
 
-	root, cmdName, ok := strings.Cut(base, "_")
-	if !ok { // no `_`, so not a subcommand
-		cmdName = root
-	}
+		root, cmdName, ok := strings.Cut(base, "_")
+		if !ok { // no `_`, so not a subcommand
+			cmdName = root
+		}
+
+		cmdName = strings.ReplaceAll(cmdName, "_", " ")
 
-	return fmt.Sprintf(frontmatter, cmdName, root)
+		return fmt.Sprintf(frontmatter, cmdName, root, sha)
+	}
 }
 
 func linkHandler(filename string) string {
diff --git a/go/cmd/vtcombo/main.go b/go/cmd/vtcombo/main.go
index acbea8ff49..ff52c28421 100644
--- a/go/cmd/vtcombo/main.go
+++ b/go/cmd/vtcombo/main.go
@@ -346,3 +346,8 @@ func (mysqld *vtcomboMysqld) StopReplication(hookExtraEnv map[string]string) err
 func (mysqld *vtcomboMysqld) SetSemiSyncEnabled(source, replica bool) error {
 	return nil
 }
+
+// SemiSyncExtensionLoaded implements the MysqlDaemon interface
+func (mysqld *vtcomboMysqld) SemiSyncExtensionLoaded() (bool, error) {
+	return true, nil
+}
diff --git a/go/vt/mysqlctl/fakemysqldaemon/fakemysqldaemon.go b/go/vt/mysqlctl/fakemysqldaemon/fakemysqldaemon.go
index 5831d817d4..3effa9309c 100644
--- a/go/vt/mysqlctl/fakemysqldaemon/fakemysqldaemon.go
+++ b/go/vt/mysqlctl/fakemysqldaemon/fakemysqldaemon.go
@@ -642,6 +642,11 @@ func (fmd *FakeMysqlDaemon) SemiSyncClients() uint32 {
 	return 0
 }
 
+// SemiSyncExtensionLoaded is part of the MysqlDaemon interface.
+func (fmd *FakeMysqlDaemon) SemiSyncExtensionLoaded() (bool, error) {
+	return true, nil
+}
+
 // SemiSyncSettings is part of the MysqlDaemon interface.
 func (fmd *FakeMysqlDaemon) SemiSyncSettings() (timeout uint64, numReplicas uint32) {
 	return 10000000, 1
diff --git a/go/vt/mysqlctl/mysql_daemon.go b/go/vt/mysqlctl/mysql_daemon.go
index c4c76224b3..ec96eee7b2 100644
--- a/go/vt/mysqlctl/mysql_daemon.go
+++ b/go/vt/mysqlctl/mysql_daemon.go
@@ -57,6 +57,7 @@ type MysqlDaemon interface {
 	GetGTIDPurged(ctx context.Context) (mysql.Position, error)
 	SetSemiSyncEnabled(source, replica bool) error
 	SemiSyncEnabled() (source, replica bool)
+	SemiSyncExtensionLoaded() (bool, error)
 	SemiSyncStatus() (source, replica bool)
 	SemiSyncClients() (count uint32)
 	SemiSyncSettings() (timeout uint64, numReplicas uint32)
diff --git a/go/vt/mysqlctl/replication.go b/go/vt/mysqlctl/replication.go
index 670a9b4e1e..3c866019c6 100644
--- a/go/vt/mysqlctl/replication.go
+++ b/go/vt/mysqlctl/replication.go
@@ -656,3 +656,16 @@ func (mysqld *Mysqld) SemiSyncReplicationStatus() (bool, error) {
 	}
 	return false, nil
 }
+
+// SemiSyncExtensionLoaded returns whether semi-sync plugins are loaded.
+func (mysqld *Mysqld) SemiSyncExtensionLoaded() (bool, error) {
+	qr, err := mysqld.FetchSuperQuery(context.Background(), "SELECT COUNT(*) > 0 AS plugin_loaded FROM information_schema.plugins WHERE plugin_name LIKE 'rpl_semi_sync%'")
+	if err != nil {
+		return false, err
+	}
+	pluginPresent, err := qr.Rows[0][0].ToBool()
+	if err != nil {
+		return false, err
+	}
+	return pluginPresent, nil
+}
diff --git a/go/vt/schema/ddl_strategy.go b/go/vt/schema/ddl_strategy.go
index 05e2e15c1f..26d1878c95 100644
--- a/go/vt/schema/ddl_strategy.go
+++ b/go/vt/schema/ddl_strategy.go
@@ -19,12 +19,15 @@ package schema
 import (
 	"fmt"
 	"regexp"
+	"strconv"
+	"time"
 
 	"github.com/google/shlex"
 )
 
 var (
-	strategyParserRegexp = regexp.MustCompile(`^([\S]+)\s+(.*)$`)
+	strategyParserRegexp      = regexp.MustCompile(`^([\S]+)\s+(.*)$`)
+	retainArtifactsFlagRegexp = regexp.MustCompile(fmt.Sprintf(`^[-]{1,2}%s=(.*?)$`, retainArtifactsFlag))
 )
 
 const (
@@ -38,6 +41,7 @@ const (
 	allowConcurrentFlag    = "allow-concurrent"
 	fastOverRevertibleFlag = "fast-over-revertible"
 	fastRangeRotationFlag  = "fast-range-rotation"
+	retainArtifactsFlag    = "retain-artifacts"
 	vreplicationTestSuite  = "vreplication-test-suite"
 )
 
@@ -98,6 +102,9 @@ func ParseDDLStrategy(strategyVariable string) (*DDLStrategySetting, error) {
 	default:
 		return nil, fmt.Errorf("Unknown online DDL strategy: '%v'", strategy)
 	}
+	if _, err := setting.RetainArtifactsDuration(); err != nil {
+		return nil, err
+	}
 	return setting, nil
 }
 
@@ -168,7 +175,34 @@ func (setting *DDLStrategySetting) IsFastRangeRotationFlag() bool {
 	return setting.hasFlag(fastRangeRotationFlag)
 }
 
-// IsVreplicationTestSuite checks if strategy options include -vreplicatoin-test-suite
+// isRetainArtifactsFlag returns true when given option denotes a `--retain-artifacts=[...]` flag
+func isRetainArtifactsFlag(opt string) (string, bool) {
+	submatch := retainArtifactsFlagRegexp.FindStringSubmatch(opt)
+	if len(submatch) == 0 {
+		return "", false
+	}
+	return submatch[1], true
+}
+
+// RetainArtifactsDuration returns a the duration indicated by --retain-artifacts
+func (setting *DDLStrategySetting) RetainArtifactsDuration() (d time.Duration, err error) {
+	// We do some ugly manual parsing of --retain-artifacts
+	opts, _ := shlex.Split(setting.Options)
+	for _, opt := range opts {
+		if val, isRetainArtifacts := isRetainArtifactsFlag(opt); isRetainArtifacts {
+			// value is possibly quoted
+			if s, err := strconv.Unquote(val); err == nil {
+				val = s
+			}
+			if val != "" {
+				d, err = time.ParseDuration(val)
+			}
+		}
+	}
+	return d, err
+}
+
+// IsVreplicationTestSuite checks if strategy options include --vreplicatoin-test-suite
 func (setting *DDLStrategySetting) IsVreplicationTestSuite() bool {
 	return setting.hasFlag(vreplicationTestSuite)
 }
@@ -178,6 +212,9 @@ func (setting *DDLStrategySetting) RuntimeOptions() []string {
 	opts, _ := shlex.Split(setting.Options)
 	validOpts := []string{}
 	for _, opt := range opts {
+		if _, ok := isRetainArtifactsFlag(opt); ok {
+			continue
+		}
 		switch {
 		case isFlag(opt, declarativeFlag):
 		case isFlag(opt, skipTopoFlag):
diff --git a/go/vt/vtgate/engine/join.go b/go/vt/vtgate/engine/join.go
index c67a0951b3..c47b523b9b 100644
--- a/go/vt/vtgate/engine/join.go
+++ b/go/vt/vtgate/engine/join.go
@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"vitess.io/vitess/go/sqltypes"
+	"vitess.io/vitess/go/sync2"
 	querypb "vitess.io/vitess/go/vt/proto/query"
 )
 
@@ -95,34 +96,36 @@ func (jn *Join) TryExecute(ctx context.Context, vcursor VCursor, bindVars map[st
 
 // TryStreamExecute performs a streaming exec.
 func (jn *Join) TryStreamExecute(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
-	joinVars := make(map[string]*querypb.BindVariable)
-	err := vcursor.StreamExecutePrimitive(ctx, jn.Left, bindVars, wantfields, func(lresult *sqltypes.Result) error {
+	var fieldNeeded sync2.AtomicBool
+	fieldNeeded.Set(wantfields)
+	err := vcursor.StreamExecutePrimitive(ctx, jn.Left, bindVars, fieldNeeded.Get(), func(lresult *sqltypes.Result) error {
+		joinVars := make(map[string]*querypb.BindVariable)
 		for _, lrow := range lresult.Rows {
 			for k, col := range jn.Vars {
 				joinVars[k] = sqltypes.ValueBindVariable(lrow[col])
 			}
-			rowSent := false
-			err := vcursor.StreamExecutePrimitive(ctx, jn.Right, combineVars(bindVars, joinVars), wantfields, func(rresult *sqltypes.Result) error {
+			var rowSent sync2.AtomicBool
+			err := vcursor.StreamExecutePrimitive(ctx, jn.Right, combineVars(bindVars, joinVars), fieldNeeded.Get(), func(rresult *sqltypes.Result) error {
 				result := &sqltypes.Result{}
-				if wantfields {
+				if fieldNeeded.Get() {
 					// This code is currently unreachable because the first result
 					// will always be just the field info, which will cause the outer
 					// wantfields code path to be executed. But this may change in the future.
-					wantfields = false
+					fieldNeeded.Set(false)
 					result.Fields = joinFields(lresult.Fields, rresult.Fields, jn.Cols)
 				}
 				for _, rrow := range rresult.Rows {
 					result.Rows = append(result.Rows, joinRows(lrow, rrow, jn.Cols))
 				}
 				if len(rresult.Rows) != 0 {
-					rowSent = true
+					rowSent.Set(true)
 				}
 				return callback(result)
 			})
 			if err != nil {
 				return err
 			}
-			if jn.Opcode == LeftJoin && !rowSent {
+			if jn.Opcode == LeftJoin && !rowSent.Get() {
 				result := &sqltypes.Result{}
 				result.Rows = [][]sqltypes.Value{joinRows(
 					lrow,
@@ -132,8 +135,8 @@ func (jn *Join) TryStreamExecute(ctx context.Context, vcursor VCursor, bindVars
 				return callback(result)
 			}
 		}
-		if wantfields {
-			wantfields = false
+		if fieldNeeded.Get() {
+			fieldNeeded.Set(false)
 			for k := range jn.Vars {
 				joinVars[k] = sqltypes.NullBindVariable
 			}
diff --git a/go/vt/vttablet/onlineddl/executor.go b/go/vt/vttablet/onlineddl/executor.go
index d64ee55e7b..dfd4525d60 100644
--- a/go/vt/vttablet/onlineddl/executor.go
+++ b/go/vt/vttablet/onlineddl/executor.go
@@ -4527,6 +4527,11 @@ func (e *Executor) SubmitMigration(
 
 	revertedUUID, _ := onlineDDL.GetRevertUUID() // Empty value if the migration is not actually a REVERT. Safe to ignore error.
 	retainArtifactsSeconds := int64((retainOnlineDDLTables).Seconds())
+	if retainArtifacts, _ := onlineDDL.StrategySetting().RetainArtifactsDuration(); retainArtifacts != 0 {
+		// Explicit retention indicated by `--retain-artifact` DDL strategy flag for this migration. Override!
+		retainArtifactsSeconds = int64((retainArtifacts).Seconds())
+	}
+
 	_, allowConcurrentMigration := e.allowConcurrentMigration(onlineDDL)
 	query, err := sqlparser.ParseAndBind(sqlInsertMigration,
 		sqltypes.StringBindVariable(onlineDDL.UUID),
diff --git a/go/vt/vttablet/onlineddl/schema.go b/go/vt/vttablet/onlineddl/schema.go
index 833148939b..276eb19b73 100644
--- a/go/vt/vttablet/onlineddl/schema.go
+++ b/go/vt/vttablet/onlineddl/schema.go
@@ -404,7 +404,7 @@ const (
 			log_path
 		FROM _vt.schema_migrations
 		WHERE
-			migration_status IN ('complete', 'failed')
+			migration_status IN ('complete', 'cancelled', 'failed')
 			AND cleanup_timestamp IS NULL
 			AND completed_timestamp <= IF(retain_artifacts_seconds=0,
 				NOW() - INTERVAL %a SECOND,
diff --git a/go/vt/vttablet/tabletmanager/rpc_actions.go b/go/vt/vttablet/tabletmanager/rpc_actions.go
index 1093c331a1..d51f5d2f84 100644
--- a/go/vt/vttablet/tabletmanager/rpc_actions.go
+++ b/go/vt/vttablet/tabletmanager/rpc_actions.go
@@ -30,6 +30,7 @@ import (
 
 	tabletmanagerdatapb "vitess.io/vitess/go/vt/proto/tabletmanagerdata"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 )
 
 // DBAction is used to tell ChangeTabletType whether to call SetReadOnly on change to
@@ -82,7 +83,13 @@ func (tm *TabletManager) ChangeType(ctx context.Context, tabletType topodatapb.T
 		return err
 	}
 	defer tm.unlock()
-	return tm.changeTypeLocked(ctx, tabletType, DBActionNone, convertBoolToSemiSyncAction(semiSync))
+
+	semiSyncAction, err := tm.convertBoolToSemiSyncAction(semiSync)
+	if err != nil {
+		return err
+	}
+
+	return tm.changeTypeLocked(ctx, tabletType, DBActionNone, semiSyncAction)
 }
 
 // ChangeType changes the tablet type
@@ -142,9 +149,23 @@ func (tm *TabletManager) RunHealthCheck(ctx context.Context) {
 	tm.QueryServiceControl.BroadcastHealth()
 }
 
-func convertBoolToSemiSyncAction(semiSync bool) SemiSyncAction {
-	if semiSync {
-		return SemiSyncActionSet
+func (tm *TabletManager) convertBoolToSemiSyncAction(semiSync bool) (SemiSyncAction, error) {
+	semiSyncExtensionLoaded, err := tm.MysqlDaemon.SemiSyncExtensionLoaded()
+	if err != nil {
+		return SemiSyncActionNone, err
+	}
+
+	if semiSyncExtensionLoaded {
+		if semiSync {
+			return SemiSyncActionSet, nil
+		} else {
+			return SemiSyncActionUnset, nil
+		}
+	} else {
+		if semiSync {
+			return SemiSyncActionNone, vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, "semi-sync plugins are not loaded")
+		} else {
+			return SemiSyncActionNone, nil
+		}
 	}
-	return SemiSyncActionUnset
 }
diff --git a/go/vt/vttablet/tabletmanager/rpc_replication.go b/go/vt/vttablet/tabletmanager/rpc_replication.go
index 55fbd8c195..79f6ef7361 100644
--- a/go/vt/vttablet/tabletmanager/rpc_replication.go
+++ b/go/vt/vttablet/tabletmanager/rpc_replication.go
@@ -294,7 +294,12 @@ func (tm *TabletManager) StartReplication(ctx context.Context, semiSync bool) er
 		}
 	}()
 
-	if err := tm.fixSemiSync(tm.Tablet().Type, convertBoolToSemiSyncAction(semiSync)); err != nil {
+	semiSyncAction, err := tm.convertBoolToSemiSyncAction(semiSync)
+	if err != nil {
+		return err
+	}
+
+	if err := tm.fixSemiSync(tm.Tablet().Type, semiSyncAction); err != nil {
 		return err
 	}
 	return tm.MysqlDaemon.StartReplication(tm.hookExtraEnv())
@@ -377,16 +382,21 @@ func (tm *TabletManager) InitPrimary(ctx context.Context, semiSync bool) (string
 		return "", err
 	}
 
+	semiSyncAction, err := tm.convertBoolToSemiSyncAction(semiSync)
+	if err != nil {
+		return "", err
+	}
+
 	// Set the server read-write, from now on we can accept real
 	// client writes. Note that if semi-sync replication is enabled,
 	// we'll still need some replicas to be able to commit transactions.
-	if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_PRIMARY, DBActionSetReadWrite, convertBoolToSemiSyncAction(semiSync)); err != nil {
+	if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_PRIMARY, DBActionSetReadWrite, semiSyncAction); err != nil {
 		return "", err
 	}
 
 	// Enforce semi-sync after changing the tablet)type to PRIMARY. Otherwise, the
 	// primary will hang while trying to create the database.
-	if err := tm.fixSemiSync(topodatapb.TabletType_PRIMARY, convertBoolToSemiSyncAction(semiSync)); err != nil {
+	if err := tm.fixSemiSync(topodatapb.TabletType_PRIMARY, semiSyncAction); err != nil {
 		return "", err
 	}
 
@@ -423,11 +433,16 @@ func (tm *TabletManager) InitReplica(ctx context.Context, parent *topodatapb.Tab
 	}
 	defer tm.unlock()
 
+	semiSyncAction, err := tm.convertBoolToSemiSyncAction(semiSync)
+	if err != nil {
+		return err
+	}
+
 	// If we were a primary type, switch our type to replica.  This
 	// is used on the old primary when using InitShardPrimary with
 	// -force, and the new primary is different from the old primary.
 	if tm.Tablet().Type == topodatapb.TabletType_PRIMARY {
-		if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_REPLICA, DBActionNone, convertBoolToSemiSyncAction(semiSync)); err != nil {
+		if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_REPLICA, DBActionNone, semiSyncAction); err != nil {
 			return err
 		}
 	}
@@ -450,7 +465,7 @@ func (tm *TabletManager) InitReplica(ctx context.Context, parent *topodatapb.Tab
 	if tt == topodatapb.TabletType_PRIMARY {
 		tt = topodatapb.TabletType_REPLICA
 	}
-	if err := tm.fixSemiSync(tt, convertBoolToSemiSyncAction(semiSync)); err != nil {
+	if err := tm.fixSemiSync(tt, semiSyncAction); err != nil {
 		return err
 	}
 
@@ -601,8 +616,13 @@ func (tm *TabletManager) UndoDemotePrimary(ctx context.Context, semiSync bool) e
 	}
 	defer tm.unlock()
 
+	semiSyncAction, err := tm.convertBoolToSemiSyncAction(semiSync)
+	if err != nil {
+		return err
+	}
+
 	// If using semi-sync, we need to enable source-side.
-	if err := tm.fixSemiSync(topodatapb.TabletType_PRIMARY, convertBoolToSemiSyncAction(semiSync)); err != nil {
+	if err := tm.fixSemiSync(topodatapb.TabletType_PRIMARY, semiSyncAction); err != nil {
 		return err
 	}
 
@@ -670,9 +690,14 @@ func (tm *TabletManager) SetReplicationSource(ctx context.Context, parentAlias *
 	}
 	defer tm.unlock()
 
+	semiSyncAction, err := tm.convertBoolToSemiSyncAction(semiSync)
+	if err != nil {
+		return err
+	}
+
 	// setReplicationSourceLocked also fixes the semi-sync. In case the tablet type is primary it assumes that it will become a replica if SetReplicationSource
 	// is called, so we always call fixSemiSync with a non-primary tablet type. This will always set the source side replication to false.
-	return tm.setReplicationSourceLocked(ctx, parentAlias, timeCreatedNS, waitPosition, forceStartReplication, convertBoolToSemiSyncAction(semiSync))
+	return tm.setReplicationSourceLocked(ctx, parentAlias, timeCreatedNS, waitPosition, forceStartReplication, semiSyncAction)
 }
 
 func (tm *TabletManager) setReplicationSourceRepairReplication(ctx context.Context, parentAlias *topodatapb.TabletAlias, timeCreatedNS int64, waitPosition string, forceStartReplication bool) (err error) {
@@ -962,8 +987,13 @@ func (tm *TabletManager) PromoteReplica(ctx context.Context, semiSync bool) (str
 		return "", err
 	}
 
+	semiSyncAction, err := tm.convertBoolToSemiSyncAction(semiSync)
+	if err != nil {
+		return "", err
+	}
+
 	// If using semi-sync, we need to enable it before going read-write.
-	if err := tm.fixSemiSync(topodatapb.TabletType_PRIMARY, convertBoolToSemiSyncAction(semiSync)); err != nil {
+	if err := tm.fixSemiSync(topodatapb.TabletType_PRIMARY, semiSyncAction); err != nil {
 		return "", err
 	}
 
diff --git a/go/vt/vttablet/tabletmanager/tm_init.go b/go/vt/vttablet/tabletmanager/tm_init.go
index cf1ea9f3d6..f4e1f70279 100644
--- a/go/vt/vttablet/tabletmanager/tm_init.go
+++ b/go/vt/vttablet/tabletmanager/tm_init.go
@@ -909,8 +909,15 @@ func (tm *TabletManager) initializeReplication(ctx context.Context, tabletType t
 	}
 	// If using semi-sync, we need to enable it before connecting to primary.
 	// We should set the correct type, since it is used in replica semi-sync
+
 	tablet.Type = tabletType
-	if err := tm.fixSemiSync(tabletType, convertBoolToSemiSyncAction(reparentutil.IsReplicaSemiSync(durability, currentPrimary.Tablet, tablet))); err != nil {
+
+	semiSyncAction, err := tm.convertBoolToSemiSyncAction(reparentutil.IsReplicaSemiSync(durability, currentPrimary.Tablet, tablet))
+	if err != nil {
+		return nil, err
+	}
+
+	if err := tm.fixSemiSync(tabletType, semiSyncAction); err != nil {
 		return nil, err
 	}
 
diff --git a/go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go b/go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go
index d4a74540d3..2204e8f18a 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go
@@ -418,9 +418,28 @@ func (tpb *tablePlanBuilder) analyzeExpr(selExpr sqlparser.SelectExpr) (*colExpr
 		references: make(map[string]bool),
 	}
 	if expr, ok := aliased.Expr.(*sqlparser.ConvertUsingExpr); ok {
+		// Here we find the actual column name in the convert, in case
+		// this is a column rename and the AS is the new column.
+		// For example, in convert(c1 using utf8mb4) as c2, we want to find
+		// c1, because c1 exists in the current table whereas c2 is the renamed column
+		// in the desired table.
+		var colName sqlparser.IdentifierCI
+		err := sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
+			switch node := node.(type) {
+			case *sqlparser.ColName:
+				if !node.Qualifier.IsEmpty() {
+					return false, fmt.Errorf("unsupported qualifier for column: %v", sqlparser.String(node))
+				}
+				colName = node.Name
+			}
+			return true, nil
+		}, aliased.Expr)
+		if err != nil {
+			return nil, fmt.Errorf("failed to find column name for convert using expression: %v, %v", sqlparser.String(aliased.Expr), err)
+		}
 		selExpr := &sqlparser.ConvertUsingExpr{
 			Type: "utf8mb4",
-			Expr: &sqlparser.ColName{Name: as},
+			Expr: &sqlparser.ColName{Name: colName},
 		}
 		cexpr.expr = expr
 		cexpr.operation = opExpr
diff --git a/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go b/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
index d45dceda1b..0378d04c37 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
@@ -698,7 +698,26 @@ func (plan *Plan) analyzeExpr(vschema *localVSchema, selExpr sqlparser.SelectExp
 			FixedValue: sqltypes.NewInt64(num),
 		}, nil
 	case *sqlparser.ConvertUsingExpr:
-		colnum, err := findColumn(plan.Table, aliased.As)
+		// Here we find the actual column name in the convert, in case
+		// this is a column rename and the AS is the new column.
+		// For example, in convert(c1 using utf8mb4) as c2, we want to find
+		// c1, because c1 exists in the current table whereas c2 is the renamed column
+		// in the desired table.
+		var colName sqlparser.IdentifierCI
+		err := sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
+			switch node := node.(type) {
+			case *sqlparser.ColName:
+				if !node.Qualifier.IsEmpty() {
+					return false, fmt.Errorf("unsupported qualifier for column: %v", sqlparser.String(node))
+				}
+				colName = node.Name
+			}
+			return true, nil
+		}, aliased.Expr)
+		if err != nil {
+			return ColExpr{}, fmt.Errorf("failed to find column name for convert using expression: %v, %v", sqlparser.String(aliased.Expr), err)
+		}
+		colnum, err := findColumn(plan.Table, colName)
 		if err != nil {
 			return ColExpr{}, err
 		}


Test patch:
diff --git a/.github/workflows/upgrade_downgrade_test_backups_e2e_next_release.yml b/.github/workflows/upgrade_downgrade_test_backups_e2e_next_release.yml
index c89b012430..15648ed60f 100644
--- a/.github/workflows/upgrade_downgrade_test_backups_e2e_next_release.yml
+++ b/.github/workflows/upgrade_downgrade_test_backups_e2e_next_release.yml
@@ -79,7 +79,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@v3
       with:
-        go-version: 1.20.5
+        go-version: 1.20.8
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/.github/workflows/upgrade_downgrade_test_backups_manual_next_release.yml b/.github/workflows/upgrade_downgrade_test_backups_manual_next_release.yml
index 833dc109f3..472162156d 100644
--- a/.github/workflows/upgrade_downgrade_test_backups_manual_next_release.yml
+++ b/.github/workflows/upgrade_downgrade_test_backups_manual_next_release.yml
@@ -81,7 +81,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@v3
       with:
-        go-version: 1.20.5
+        go-version: 1.20.8
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/.github/workflows/upgrade_downgrade_test_query_serving_queries_next_release.yml b/.github/workflows/upgrade_downgrade_test_query_serving_queries_next_release.yml
index fbd5091736..23b6a646f9 100644
--- a/.github/workflows/upgrade_downgrade_test_query_serving_queries_next_release.yml
+++ b/.github/workflows/upgrade_downgrade_test_query_serving_queries_next_release.yml
@@ -81,7 +81,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@v3
       with:
-        go-version: 1.20.5
+        go-version: 1.20.8
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/.github/workflows/upgrade_downgrade_test_query_serving_schema_next_release.yml b/.github/workflows/upgrade_downgrade_test_query_serving_schema_next_release.yml
index f2ec440be6..c5075d38c6 100644
--- a/.github/workflows/upgrade_downgrade_test_query_serving_schema_next_release.yml
+++ b/.github/workflows/upgrade_downgrade_test_query_serving_schema_next_release.yml
@@ -81,7 +81,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@v3
       with:
-        go-version: 1.20.5
+        go-version: 1.20.8
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/.github/workflows/upgrade_downgrade_test_reparent_new_vtctl.yml b/.github/workflows/upgrade_downgrade_test_reparent_new_vtctl.yml
index 8cbc625c0d..c2bcae6c68 100644
--- a/.github/workflows/upgrade_downgrade_test_reparent_new_vtctl.yml
+++ b/.github/workflows/upgrade_downgrade_test_reparent_new_vtctl.yml
@@ -81,7 +81,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@v3
       with:
-        go-version: 1.20.5
+        go-version: 1.20.8
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/.github/workflows/upgrade_downgrade_test_reparent_new_vttablet.yml b/.github/workflows/upgrade_downgrade_test_reparent_new_vttablet.yml
index 95c86414c9..a24d3e5975 100644
--- a/.github/workflows/upgrade_downgrade_test_reparent_new_vttablet.yml
+++ b/.github/workflows/upgrade_downgrade_test_reparent_new_vttablet.yml
@@ -81,7 +81,7 @@ jobs:
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
       uses: actions/setup-go@v3
       with:
-        go-version: 1.20.5
+        go-version: 1.20.8
 
     - name: Set up python
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
diff --git a/go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go b/go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go
index d5d7cffab0..5275d45583 100644
--- a/go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go
+++ b/go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go
@@ -154,10 +154,8 @@ func TestMain(m *testing.M) {
 	if err != nil {
 		fmt.Printf("%v\n", err)
 		os.Exit(1)
-	} else {
-		os.Exit(exitcode)
 	}
-
+	os.Exit(exitcode)
 }
 
 func TestSchemaChange(t *testing.T) {
diff --git a/go/test/endtoend/onlineddl/vrepl_suite/testdata/rename-retype-json/after_columns b/go/test/endtoend/onlineddl/vrepl_suite/testdata/rename-retype-json/after_columns
new file mode 100644
index 0000000000..99f8609786
--- /dev/null
+++ b/go/test/endtoend/onlineddl/vrepl_suite/testdata/rename-retype-json/after_columns
@@ -0,0 +1,1 @@
+id, c1j
diff --git a/go/test/endtoend/onlineddl/vrepl_suite/testdata/rename-retype-json/alter b/go/test/endtoend/onlineddl/vrepl_suite/testdata/rename-retype-json/alter
new file mode 100644
index 0000000000..f2e64ff089
--- /dev/null
+++ b/go/test/endtoend/onlineddl/vrepl_suite/testdata/rename-retype-json/alter
@@ -0,0 +1,1 @@
+change column c1 c1j json
diff --git a/go/test/endtoend/onlineddl/vrepl_suite/testdata/rename-retype-json/before_columns b/go/test/endtoend/onlineddl/vrepl_suite/testdata/rename-retype-json/before_columns
new file mode 100644
index 0000000000..b791aa0d27
--- /dev/null
+++ b/go/test/endtoend/onlineddl/vrepl_suite/testdata/rename-retype-json/before_columns
@@ -0,0 +1,1 @@
+id, c1
diff --git a/go/test/endtoend/onlineddl/vrepl_suite/testdata/rename-retype-json/create.sql b/go/test/endtoend/onlineddl/vrepl_suite/testdata/rename-retype-json/create.sql
new file mode 100644
index 0000000000..5280498e9f
--- /dev/null
+++ b/go/test/endtoend/onlineddl/vrepl_suite/testdata/rename-retype-json/create.sql
@@ -0,0 +1,22 @@
+drop table if exists onlineddl_test;
+create table onlineddl_test (
+  id int auto_increment,
+  c1 int not null,
+  primary key (id)
+) auto_increment=1;
+
+insert into onlineddl_test values (1, 11);
+insert into onlineddl_test values (2, 13);
+
+drop event if exists onlineddl_test;
+delimiter ;;
+create event onlineddl_test
+  on schedule every 1 second
+  starts current_timestamp
+  ends current_timestamp + interval 60 second
+  on completion not preserve
+  enable
+  do
+begin
+  insert into onlineddl_test values (null, 17);
+end ;;
diff --git a/go/test/endtoend/reparent/newfeaturetest/reparent_test.go b/go/test/endtoend/reparent/newfeaturetest/reparent_test.go
index 4fb2b82753..394f9f7122 100644
--- a/go/test/endtoend/reparent/newfeaturetest/reparent_test.go
+++ b/go/test/endtoend/reparent/newfeaturetest/reparent_test.go
@@ -17,6 +17,8 @@ limitations under the License.
 package newfeaturetest
 
 import (
+	"context"
+	"fmt"
 	"strconv"
 	"testing"
 
@@ -182,3 +184,38 @@ func TestTabletRestart(t *testing.T) {
 	err := tablets[1].VttabletProcess.Setup()
 	require.NoError(t, err)
 }
+
+// Tests ensures that ChangeTabletType works even when semi-sync plugins are not loaded.
+func TestChangeTypeWithoutSemiSync(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	clusterInstance := utils.SetupReparentCluster(t, "none")
+	defer utils.TeardownCluster(clusterInstance)
+	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
+
+	ctx := context.Background()
+
+	primary, replica := tablets[0], tablets[1]
+
+	// Unload semi sync plugins
+	for _, tablet := range tablets[0:4] {
+		qr := utils.RunSQL(ctx, t, "select @@global.super_read_only", tablet)
+		result := fmt.Sprintf("%v", qr.Rows[0][0].ToString())
+		if result == "1" {
+			utils.RunSQL(ctx, t, "set global super_read_only = 0", tablet)
+		}
+
+		utils.RunSQL(ctx, t, "UNINSTALL PLUGIN rpl_semi_sync_slave;", tablet)
+		utils.RunSQL(ctx, t, "UNINSTALL PLUGIN rpl_semi_sync_master;", tablet)
+	}
+
+	utils.ValidateTopology(t, clusterInstance, true)
+	utils.CheckPrimaryTablet(t, clusterInstance, primary)
+
+	// Change replica's type to rdonly
+	err := clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", replica.Alias, "rdonly")
+	require.NoError(t, err)
+
+	// Change tablets type from rdonly back to replica
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", replica.Alias, "replica")
+	require.NoError(t, err)
+}
diff --git a/go/test/endtoend/reparent/utils/utils.go b/go/test/endtoend/reparent/utils/utils.go
index 3ea2dc04e3..ca1cec15bd 100644
--- a/go/test/endtoend/reparent/utils/utils.go
+++ b/go/test/endtoend/reparent/utils/utils.go
@@ -75,9 +75,25 @@ func SetupRangeBasedCluster(ctx context.Context, t *testing.T) *cluster.LocalPro
 	return setupCluster(ctx, t, ShardName, []string{cell1}, []int{2}, "semi_sync")
 }
 
-// TeardownCluster is used to teardown the reparent cluster
+// TeardownCluster is used to teardown the reparent cluster. When
+// run in a CI environment -- which is considered true when the
+// "CI" env variable is set to "true" -- the teardown also removes
+// the VTDATAROOT directory that was used for the test/cluster.
 func TeardownCluster(clusterInstance *cluster.LocalProcessCluster) {
+	usedRoot := clusterInstance.CurrentVTDATAROOT
 	clusterInstance.Teardown()
+	// This is always set to "true" on GitHub Actions runners:
+	// https://docs.github.com/en/actions/learn-github-actions/variables#default-environment-variables
+	ci, ok := os.LookupEnv("CI")
+	if !ok || strings.ToLower(ci) != "true" {
+		// Leave the directory in place to support local debugging.
+		return
+	}
+	// We're running in the CI, so free up disk space for any
+	// subsequent tests.
+	if err := os.RemoveAll(usedRoot); err != nil {
+		log.Errorf("Failed to remove previously used VTDATAROOT (%s): %v", usedRoot, err)
+	}
 }
 
 func setupCluster(ctx context.Context, t *testing.T, shardName string, cells []string, numTablets []int, durability string) *cluster.LocalProcessCluster {
diff --git a/go/vt/schema/ddl_strategy_test.go b/go/vt/schema/ddl_strategy_test.go
index 8a700655e5..d8723ee2c5 100644
--- a/go/vt/schema/ddl_strategy_test.go
+++ b/go/vt/schema/ddl_strategy_test.go
@@ -19,6 +19,7 @@ package schema
 import (
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -37,6 +38,69 @@ func TestIsDirect(t *testing.T) {
 	assert.True(t, DDLStrategy("something").IsDirect())
 }
 
+func TestIsExpireArtifactsFlag(t *testing.T) {
+	tt := []struct {
+		s      string
+		expect bool
+		val    string
+		d      time.Duration
+	}{
+		{
+			s: "something",
+		},
+		{
+			s: "-retain-artifacts",
+		},
+		{
+			s: "--retain-artifacts",
+		},
+		{
+			s:      "--retain-artifacts=",
+			expect: true,
+		},
+		{
+			s:      "--retain-artifacts=0",
+			expect: true,
+			val:    "0",
+			d:      0,
+		},
+		{
+			s:      "-retain-artifacts=0",
+			expect: true,
+			val:    "0",
+			d:      0,
+		},
+		{
+			s:      "--retain-artifacts=1m",
+			expect: true,
+			val:    "1m",
+			d:      time.Minute,
+		},
+		{
+			s:      `--retain-artifacts="1m"`,
+			expect: true,
+			val:    `"1m"`,
+			d:      time.Minute,
+		},
+	}
+	for _, ts := range tt {
+		t.Run(ts.s, func(t *testing.T) {
+			setting, err := ParseDDLStrategy("online " + ts.s)
+			assert.NoError(t, err)
+
+			val, isRetainArtifacts := isRetainArtifactsFlag(ts.s)
+			assert.Equal(t, ts.expect, isRetainArtifacts)
+			assert.Equal(t, ts.val, val)
+
+			if ts.expect {
+				d, err := setting.RetainArtifactsDuration()
+				assert.NoError(t, err)
+				assert.Equal(t, ts.d, d)
+			}
+		})
+	}
+}
+
 func TestParseDDLStrategy(t *testing.T) {
 	tt := []struct {
 		strategyVariable     string
@@ -49,6 +113,7 @@ func TestParseDDLStrategy(t *testing.T) {
 		isAllowConcurrent    bool
 		fastOverRevertible   bool
 		fastRangeRotation    bool
+		expireArtifacts      time.Duration
 		runtimeOptions       string
 		err                  error
 	}{
@@ -145,6 +210,13 @@ func TestParseDDLStrategy(t *testing.T) {
 			runtimeOptions:    "",
 			fastRangeRotation: true,
 		},
+		{
+			strategyVariable: "vitess --retain-artifacts=4m",
+			strategy:         DDLStrategyVitess,
+			options:          "--retain-artifacts=4m",
+			runtimeOptions:   "",
+			expireArtifacts:  4 * time.Minute,
+		},
 	}
 	for _, ts := range tt {
 		setting, err := ParseDDLStrategy(ts.strategyVariable)
@@ -166,4 +238,8 @@ func TestParseDDLStrategy(t *testing.T) {
 		_, err := ParseDDLStrategy("other")
 		assert.Error(t, err)
 	}
+	{
+		_, err := ParseDDLStrategy("online --retain-artifacts=3")
+		assert.Error(t, err)
+	}
 }
diff --git a/go/vt/vtgate/executor_select_test.go b/go/vt/vtgate/executor_select_test.go
index d060e7dfa6..b322205df8 100644
--- a/go/vt/vtgate/executor_select_test.go
+++ b/go/vt/vtgate/executor_select_test.go
@@ -3782,3 +3782,35 @@ func TestMain(m *testing.M) {
 	_flag.ParseFlagsForTest()
 	os.Exit(m.Run())
 }
+
+func TestStreamJoinQuery(t *testing.T) {
+	// Special setup: Don't use createExecutorEnv.
+	cell := "aa"
+	hc := discovery.NewFakeHealthCheck(nil)
+	u := createSandbox(KsTestUnsharded)
+	s := createSandbox(KsTestSharded)
+	s.VSchema = executorVSchema
+	u.VSchema = unshardedVSchema
+	serv := newSandboxForCells([]string{cell})
+	resolver := newTestResolver(hc, serv, cell)
+	shards := []string{"-20", "20-40", "40-60", "60-80", "80-a0", "a0-c0", "c0-e0", "e0-"}
+	for _, shard := range shards {
+		_ = hc.AddTestTablet(cell, shard, 1, "TestExecutor", shard, topodatapb.TabletType_PRIMARY, true, 1, nil)
+	}
+	executor := createExecutor(serv, cell, resolver)
+
+	sql := "select u.foo, u.apa, ue.bar, ue.apa from user u join user_extra ue on u.foo = ue.bar"
+	result, err := executorStream(executor, sql)
+	require.NoError(t, err)
+	wantResult := &sqltypes.Result{
+		Fields: append(sandboxconn.SingleRowResult.Fields, sandboxconn.SingleRowResult.Fields...),
+	}
+	wantRow := append(sandboxconn.StreamRowResult.Rows[0], sandboxconn.StreamRowResult.Rows[0]...)
+	for i := 0; i < 64; i++ {
+		wantResult.Rows = append(wantResult.Rows, wantRow)
+	}
+	require.Equal(t, len(wantResult.Rows), len(result.Rows))
+	for idx := 0; idx < 64; idx++ {
+		utils.MustMatch(t, wantResult.Rows[idx], result.Rows[idx], "mismatched on: ", strconv.Itoa(idx))
+	}
+}
diff --git a/go/vt/vtgate/planbuilder/testdata/from_cases.json b/go/vt/vtgate/planbuilder/testdata/from_cases.json
index 30e92c521e..71e20866a0 100644
--- a/go/vt/vtgate/planbuilder/testdata/from_cases.json
+++ b/go/vt/vtgate/planbuilder/testdata/from_cases.json
@@ -6365,5 +6365,90 @@
         "zlookup_unique.t1"
       ]
     }
+  },
+  {
+    "comment": "join query using table with muticolumn vindex",
+    "query": "select 1 from multicol_tbl m1 join multicol_tbl m2 on m1.cola = m2.cola",
+    "v3-plan": {
+      "QueryType": "SELECT",
+      "Original": "select 1 from multicol_tbl m1 join multicol_tbl m2 on m1.cola = m2.cola",
+      "Instructions": {
+        "OperatorType": "Join",
+        "Variant": "Join",
+        "JoinColumnIndexes": "L:0",
+        "JoinVars": {
+          "m1_cola": 1
+        },
+        "TableName": "multicol_tbl_multicol_tbl",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "Scatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select 1, m1.cola from multicol_tbl as m1 where 1 != 1",
+            "Query": "select 1, m1.cola from multicol_tbl as m1",
+            "Table": "multicol_tbl"
+          },
+          {
+            "OperatorType": "Route",
+            "Variant": "Scatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select 1 from multicol_tbl as m2 where 1 != 1",
+            "Query": "select 1 from multicol_tbl as m2 where m2.cola = :m1_cola",
+            "Table": "multicol_tbl"
+          }
+        ]
+      }
+    },
+    "gen4-plan": {
+      "QueryType": "SELECT",
+      "Original": "select 1 from multicol_tbl m1 join multicol_tbl m2 on m1.cola = m2.cola",
+      "Instructions": {
+        "OperatorType": "Join",
+        "Variant": "Join",
+        "JoinColumnIndexes": "L:1",
+        "JoinVars": {
+          "m1_cola": 0
+        },
+        "TableName": "multicol_tbl_multicol_tbl",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "Scatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select m1.cola, 1 from multicol_tbl as m1 where 1 != 1",
+            "Query": "select m1.cola, 1 from multicol_tbl as m1",
+            "Table": "multicol_tbl"
+          },
+          {
+            "OperatorType": "Route",
+            "Variant": "SubShard",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select 1 from multicol_tbl as m2 where 1 != 1",
+            "Query": "select 1 from multicol_tbl as m2 where m2.cola = :m1_cola",
+            "Table": "multicol_tbl",
+            "Values": [
+              ":m1_cola"
+            ],
+            "Vindex": "multicolIdx"
+          }
+        ]
+      },
+      "TablesUsed": [
+        "user.multicol_tbl"
+      ]
+    }
   }
 ]
diff --git a/go/vt/vttablet/tabletmanager/vreplication/replicator_plan_test.go b/go/vt/vttablet/tabletmanager/vreplication/replicator_plan_test.go
index 8b97f02dc1..780b1c0d06 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/replicator_plan_test.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/replicator_plan_test.go
@@ -414,6 +414,55 @@ func TestBuildPlayerPlan(t *testing.T) {
 				},
 			},
 		},
+	}, {
+		input: &binlogdatapb.Filter{
+			Rules: []*binlogdatapb.Rule{{
+				Match:  "t1",
+				Filter: "select c1, convert(c using utf8mb4) as c2 from t1",
+			}},
+		},
+		plan: &TestReplicatorPlan{
+			VStreamFilter: &binlogdatapb.Filter{
+				Rules: []*binlogdatapb.Rule{{
+					Match:  "t1",
+					Filter: "select c1, convert(c using utf8mb4) as c2 from t1",
+				}},
+			},
+			TargetTables: []string{"t1"},
+			TablePlans: map[string]*TestTablePlan{
+				"t1": {
+					TargetName:   "t1",
+					SendRule:     "t1",
+					PKReferences: []string{"c1"},
+					InsertFront:  "insert into t1(c1,c2)",
+					InsertValues: "(:a_c1,convert(:a_c using utf8mb4))",
+					Insert:       "insert into t1(c1,c2) values (:a_c1,convert(:a_c using utf8mb4))",
+					Update:       "update t1 set c2=convert(:a_c using utf8mb4) where c1=:b_c1",
+					Delete:       "delete from t1 where c1=:b_c1",
+				},
+			},
+		},
+		planpk: &TestReplicatorPlan{
+			VStreamFilter: &binlogdatapb.Filter{
+				Rules: []*binlogdatapb.Rule{{
+					Match:  "t1",
+					Filter: "select c1, convert(c using utf8mb4) as c2, pk1, pk2 from t1",
+				}},
+			},
+			TargetTables: []string{"t1"},
+			TablePlans: map[string]*TestTablePlan{
+				"t1": {
+					TargetName:   "t1",
+					SendRule:     "t1",
+					PKReferences: []string{"c1", "pk1", "pk2"},
+					InsertFront:  "insert into t1(c1,c2)",
+					InsertValues: "(:a_c1,convert(:a_c using utf8mb4))",
+					Insert:       "insert into t1(c1,c2) select :a_c1, convert(:a_c using utf8mb4) from dual where (:a_pk1,:a_pk2) <= (1,'aaa')",
+					Update:       "update t1 set c2=convert(:a_c using utf8mb4) where c1=:b_c1 and (:b_pk1,:b_pk2) <= (1,'aaa')",
+					Delete:       "delete from t1 where c1=:b_c1 and (:b_pk1,:b_pk2) <= (1,'aaa')",
+				},
+			},
+		},
 	}, {
 		// Keywords as names.
 		input: &binlogdatapb.Filter{
diff --git a/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go b/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go
index 4ec57b15b7..d1caac6370 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go
@@ -427,6 +427,25 @@ func TestPlanBuilder(t *testing.T) {
 				KeyRange:      nil,
 			}},
 		},
+	}, {
+		inTable: t1,
+		inRule:  &binlogdatapb.Rule{Match: "t1", Filter: "select convert(val using utf8mb4) as val2, id as id from t1"},
+		outPlan: &Plan{
+			ColExprs: []ColExpr{{
+				ColNum: 1,
+				Field: &querypb.Field{
+					Name: "val",
+					Type: sqltypes.VarBinary,
+				},
+			}, {
+				ColNum: 0,
+				Field: &querypb.Field{
+					Name: "id",
+					Type: sqltypes.Int64,
+				},
+			}},
+			convertUsingUTF8Columns: map[string]bool{"val": true},
+		},
 	}, {
 		inTable: regional,
 		inRule:  &binlogdatapb.Rule{Match: "regional", Filter: "select id, keyspace_id() from regional"},