Repository: vitessio/vitess
Pull request #10002 (2022-03-29 14:42:07)

Fix Flaky TestTopoCustomRule


<!--
  Thank you for your contribution to the Vitess project.
  How to contribute: https://vitess.io/docs/contributing/
  Please first make sure there is an open Issue to discuss the feature/fix suggested in this PR.
  If this is a new feature, please mark the Issue as "RFC".
 -->

<!-- if this PR is Work in Progress please create it as a Draft Pull Request -->

## Description
<!-- A few sentences describing the overall goals of the pull request's commits. -->
<!-- If this is a bug fix and you think the fix should be backported, please write so. -->
This PR fixes the flakiness of the `TestTopoCustomRule`.

## Related Issue(s)
<!-- List related issues and pull requests. If this PR fixes an issue, please add it using Fixes #????  -->
#9946 

## Checklist
- [x] Should this PR be backported? No
- [x] Tests were added or are not required
- [x] Documentation was added or is not required

## Deployment Notes
<!-- Notes regarding deployment of the contained body of work. These should note any db migrations, etc. -->


Patch:
diff --git a/Makefile b/Makefile
index b80cd1a108..c2df1b5f03 100644
--- a/Makefile
+++ b/Makefile
@@ -87,7 +87,12 @@ endif
 	# For the specified GOOS + GOARCH, build all the binaries by default with CGO disabled
 	CGO_ENABLED=0 GOOS=${GOOS} GOARCH=${GOARCH} go install -trimpath $(EXTRA_BUILD_FLAGS) $(VT_GO_PARALLEL) -ldflags "$(shell tools/build_version_flags.sh)" ./go/...
 	# unset GOOS and embed local resources in the vttablet executable
-	(cd go/cmd/vttablet && unset GOOS && unset GOARCH && go run github.com/GeertJohan/go.rice/rice --verbose append --exec=$${HOME}/go/bin/${GOOS}_${GOARCH}/vttablet)
+	if [ -d /go/bin ]; then
+		# Probably in the bootstrap container
+		(cd go/cmd/vttablet && go run github.com/GeertJohan/go.rice/rice --verbose append --exec=/go/bin/${GOOS}_${GOARCH}/vttablet)
+	else
+		(cd go/cmd/vttablet && unset GOOS && unset GOARCH && go run github.com/GeertJohan/go.rice/rice --verbose append --exec=$${HOME}/go/bin/${GOOS}_${GOARCH}/vttablet)
+	fi
 	# Cross-compiling w/ cgo isn't trivial and we don't need vtorc, so we can skip building it
 
 debug:
@@ -105,6 +110,13 @@ install: build
 	mkdir -p "$${PREFIX}/bin"
 	cp "$${VTROOT}/bin/"{mysqlctl,mysqlctld,vtorc,vtadmin,vtctld,vtctlclient,vtctldclient,vtgate,vttablet,vtworker,vtbackup} "$${PREFIX}/bin/"
 
+# Will only work inside the docker bootstrap for now
+cross-install: cross-build
+	# binaries
+	mkdir -p "$${PREFIX}/bin"
+	# Still no vtorc for cross-compile
+	cp "/go/bin/${GOOS}_${GOARCH}/"{mysqlctl,mysqlctld,vtadmin,vtctld,vtctlclient,vtctldclient,vtgate,vttablet,vtworker,vtbackup} "$${PREFIX}/bin/"
+
 # Install local install the binaries needed to run vitess locally
 # Usage: make install-local PREFIX=/path/to/install/root
 install-local: build
@@ -269,7 +281,14 @@ define build_docker_image
 	${info Building ${2}}
 	# Fix permissions before copying files, to avoid AUFS bug other must have read/access permissions
 	chmod -R o=rx *;
-	docker build -f ${1} -t ${2} --build-arg bootstrap_version=${BOOTSTRAP_VERSION} .;
+
+	if grep -q arm64 <<< ${2}; then \
+		echo "Building docker using arm64 buildx"; \
+		docker buildx build --platform linux/arm64 -f ${1} -t ${2} --build-arg bootstrap_version=${BOOTSTRAP_VERSION} .; \
+	else \
+		echo "Building docker using straight docker build"; \
+		docker build -f ${1} -t ${2} --build-arg bootstrap_version=${BOOTSTRAP_VERSION} .; \
+	fi
 endef
 
 docker_base:
@@ -285,7 +304,7 @@ docker_base_all: docker_base $(DOCKER_BASE_TARGETS)
 docker_lite:
 	${call build_docker_image,docker/lite/Dockerfile,vitess/lite}
 
-DOCKER_LITE_SUFFIX = mysql56 mysql57 ubi7.mysql57 mysql80 ubi7.mysql80 mariadb mariadb103 percona percona57 ubi7.percona57 percona80 ubi7.percona80 alpine testing
+DOCKER_LITE_SUFFIX = mysql56 mysql57 ubi7.mysql57 mysql80 ubi7.mysql80 mariadb mariadb103 percona percona57 ubi7.percona57 percona80 ubi7.percona80 alpine testing ubi8.mysql80 ubi8.arm64.mysql80
 DOCKER_LITE_TARGETS = $(addprefix docker_lite_,$(DOCKER_LITE_SUFFIX))
 $(DOCKER_LITE_TARGETS): docker_lite_%:
 	${call build_docker_image,docker/lite/Dockerfile.$*,vitess/lite:$*}
diff --git a/docker/lite/Dockerfile.ubi8.arm64.mysql80 b/docker/lite/Dockerfile.ubi8.arm64.mysql80
new file mode 100644
index 0000000000..9c2fcd297a
--- /dev/null
+++ b/docker/lite/Dockerfile.ubi8.arm64.mysql80
@@ -0,0 +1,108 @@
+# Copyright 2022 The Vitess Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# NOTE: We have to build the Vitess binaries from scratch instead of sharing
+#       a base image because Docker Hub dropped the feature we relied upon to
+#       ensure images contain the right binaries.
+
+# Use a temporary layer for the build stage.
+ARG bootstrap_version=5
+ARG image="vitess/bootstrap:${bootstrap_version}-mysql80"
+
+FROM "${image}" AS builder
+
+# Allows docker builds to set the BUILD_NUMBER
+ARG BUILD_NUMBER
+
+# Re-copy sources from working tree.
+COPY --chown=vitess:vitess . /vt/src/vitess.io/vitess
+
+# Build and install Vitess in a temporary output directory.
+USER vitess
+RUN make cross-install PREFIX=/vt/install GOOS=linux GOARCH=arm64
+
+# Start over and build the final image.
+FROM registry.access.redhat.com/ubi8/ubi:latest
+
+# Install keys and dependencies
+RUN mkdir /tmp/gpg && chmod 700 /tmp/gpg && export GNUPGHOME=/tmp/gpg \
+ && yum install -y --setopt=alwaysprompt=no gnupg \
+ && ( gpg --keyserver keyserver.ubuntu.com --recv-keys 430BDF5C56E7C94E848EE60C1C4CBDCDCD2EFD2A 4D1BB29D63D98E422B2113B19334A25F8507EFA5 99DB70FAE1D7CE227FB6488205B555B38483C65D 3A79BD29 A4A9406876FCBD3C456770C88C718D3B5072E1F5 94E279EB8D8F25B21810ADF121EA45AB2F86D6A1 ) \
+ # No xtrabackup packages for aarch64 yet, but still keeping this here
+ && gpg --export --armor 430BDF5C56E7C94E848EE60C1C4CBDCDCD2EFD2A > ${GNUPGHOME}/RPM-GPG-KEY-Percona.1 \
+ && gpg --export --armor 4D1BB29D63D98E422B2113B19334A25F8507EFA5 > ${GNUPGHOME}/RPM-GPG-KEY-Percona.2 \
+ && gpg --export --armor 99DB70FAE1D7CE227FB6488205B555B38483C65D > ${GNUPGHOME}/RPM-GPG-KEY-CentOS-8 \
+ && gpg --export --armor 94E279EB8D8F25B21810ADF121EA45AB2F86D6A1 > ${GNUPGHOME}/RPM-GPG-KEY-EPEL-8 \
+ && gpg --export --armor 3A79BD29 > ${GNUPGHOME}/RPM-GPG-KEY-MySQL.1 \
+ && gpg --export --armor A4A9406876FCBD3C456770C88C718D3B5072E1F5 > ${GNUPGHOME}/RPM-GPG-KEY-MySQL.2 \
+ && rpmkeys --import ${GNUPGHOME}/RPM-GPG-KEY-Percona.1 ${GNUPGHOME}/RPM-GPG-KEY-Percona.2 ${GNUPGHOME}/RPM-GPG-KEY-CentOS-8 ${GNUPGHOME}/RPM-GPG-KEY-MySQL.1 ${GNUPGHOME}/RPM-GPG-KEY-MySQL.2 /etc/pki/rpm-gpg/RPM-GPG-KEY-redhat-release \
+ && curl -L --retry-delay 10 --retry 3 -o /tmp/mysqlrepo.rpm https://dev.mysql.com/get/mysql80-community-release-el8-1.noarch.rpm \
+ && curl -L --retry-delay 10 --retry 3 -o /tmp/perconarepo.rpm https://repo.percona.com/yum/percona-release-latest.noarch.rpm \
+ && rpmkeys --checksig /tmp/mysqlrepo.rpm /tmp/perconarepo.rpm \
+ && rpm -Uvh /tmp/mysqlrepo.rpm /tmp/perconarepo.rpm \
+ && rm -f /tmp/mysqlrepo.rpm /tmp/perconarepo.rpm
+RUN echo H4sICIDAHmICA2ZvbwDVkDFLxEAQhfv9FVfY7o4RhCBsoXJcIXKHwUIOi7m5MVk2yS6zG0//vYlRULTU4rrHvOHN+2ZL5Q4TP6oeO7bX3Od1pcuFXlyNUzVZg7S2yTmmCwDsgzjuDSUyB5SDI2+QzOChcyJBEnwkPOPQZijNuTkrigKmsHUFJ1MeCjUQEqg61tQweVtM0vOrfXItj1eAM0H0DiR2erTgbnOrV5uVvlk+6M+Kinvctby3p0ptqRziHjOnnxz3s/FnKJcxVlkYu/+k4Zcs+AvM8n3+jWW8MBc2NO6FZILUMEsoYQ76UvWI/vAGB/SOZZsCAAA= | base64 -d | gzip -dc > /etc/yum.repos.d/CentOS-Base.repo \
+ && yum install -y --setopt=alwaysprompt=no --setopt=tsflags=nodocs --enablerepo c8base --enablerepo c8updates --enablerepo c8extras libev numactl-libs sysstat strace \
+ && yum install -y --setopt=alwaysprompt=no --setopt=tsflags=nodocs https://download-ib01.fedoraproject.org/pub/epel/8/Everything/aarch64/Packages/g/gperftools-libs-2.7-9.el8.aarch64.rpm https://download-ib01.fedoraproject.org/pub/epel/8/Everything/aarch64/Packages/j/jemalloc-5.2.1-2.el8.aarch64.rpm https://download-ib01.fedoraproject.org/pub/epel/8/Everything/aarch64/Packages/l/libunwind-1.3.1-3.el8.aarch64.rpm
+RUN yum update -y --setopt=alwaysprompt=no --setopt=tsflags=nodocs \
+ && yum install -y --setopt=alwaysprompt=no --setopt=tsflags=nodocs bzip2 ca-certificates gnupg libaio libcurl \
+    procps-ng rsync wget openssl hostname curl tzdata make \
+# Can't use alwaysprompt=no here, since we need to pick up deps
+#   No way to separate key imports and accept deps separately in yum/dnf
+ && yum install -y --setopt=tsflags=nodocs --enablerepo mysql80-community \
+    mysql-community-client mysql-community-server \
+# Have to use hacks to ignore conflicts on /etc/my.cnf install
+ && mkdir -p /tmp/1 \
+# Enable xtrabackup and percona toolkit repo using the new percona-release tool, commented out for now (no aarch64 packages)
+ #&& /usr/bin/percona-release enable-only pxb-80 \
+ #&& /usr/bin/percona-release enable pt \
+ #&& yum install -y --setopt=alwaysprompt=no --downloadonly --downloaddir=/tmp/1 --enablerepo mysql80-community percona-xtrabackup-80 percona-toolkit \
+ #&& rpm -Uvh --replacefiles /tmp/1/*rpm \
+ && rm -rf /tmp/1 \
+ && yum clean all \
+ && yum clean all --enablerepo mysql80-community \
+ && rm -rf /etc/my.cnf /var/lib/mysql /tmp/gpg /sbin/mysqld-debug
+
+# Set up Vitess user and directory tree.
+RUN groupadd -g 1001 -r vitess && useradd -r -u 1001 -g vitess vitess
+RUN mkdir -p /vt/vtdataroot && chown -R vitess:vitess /vt
+
+# Set up Vitess environment (just enough to run pre-built Go binaries)
+ENV VTROOT /vt/src/vitess.io/vitess
+ENV VTDATAROOT /vt/vtdataroot
+ENV PATH $VTROOT/bin:$PATH
+ENV MYSQL_FLAVOR MySQL80
+
+# Copy artifacts from builder layer.
+COPY --from=builder --chown=vitess:vitess /vt/install /vt
+COPY --from=builder --chown=vitess:vitess /vt/src/vitess.io/vitess/web/orchestrator /vt/web/orchestrator
+
+RUN mkdir -p /licenses
+COPY LICENSE /licenses
+
+# Create mount point for actual data (e.g. MySQL data dir)
+VOLUME /vt/vtdataroot
+USER vitess
+
+LABEL name="Vitess Lite image - MySQL Community Server 8.0" \
+      io.k8s.display-name="Vitess Lite image - MySQL Community Server 8.0" \
+      maintainer="cncf-vitess-maintainers@lists.cncf.io" \
+      vendor="CNCF" \
+      version="13.0.0" \
+      release="1" \
+      summary="Vitess base container image, containing Vitess components along with MySQL Community Server 8.0" \
+      description="Vitess base container image, containing Vitess components along with MySQL Community Server 8.0" \
+      io.k8s.description="Vitess base container image, containing Vitess components along with MySQL Community Server 8.0" \
+      distribution-scope="public" \
+      url="https://vitess.io"
diff --git a/docker/lite/Dockerfile.ubi8.mysql80 b/docker/lite/Dockerfile.ubi8.mysql80
new file mode 100644
index 0000000000..5aaa8fc293
--- /dev/null
+++ b/docker/lite/Dockerfile.ubi8.mysql80
@@ -0,0 +1,106 @@
+# Copyright 2022 The Vitess Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# NOTE: We have to build the Vitess binaries from scratch instead of sharing
+#       a base image because Docker Hub dropped the feature we relied upon to
+#       ensure images contain the right binaries.
+
+# Use a temporary layer for the build stage.
+ARG bootstrap_version=5
+ARG image="vitess/bootstrap:${bootstrap_version}-mysql80"
+
+FROM "${image}" AS builder
+
+# Allows docker builds to set the BUILD_NUMBER
+ARG BUILD_NUMBER
+
+# Re-copy sources from working tree.
+COPY --chown=vitess:vitess . /vt/src/vitess.io/vitess
+
+# Build and install Vitess in a temporary output directory.
+USER vitess
+RUN make install PREFIX=/vt/install
+
+# Start over and build the final image.
+FROM registry.access.redhat.com/ubi8/ubi:latest
+
+# Install keys and dependencies
+RUN mkdir /tmp/gpg && chmod 700 /tmp/gpg && export GNUPGHOME=/tmp/gpg \
+ && yum install -y --setopt=alwaysprompt=no gnupg \
+ && ( gpg --keyserver keyserver.ubuntu.com --recv-keys 430BDF5C56E7C94E848EE60C1C4CBDCDCD2EFD2A 4D1BB29D63D98E422B2113B19334A25F8507EFA5 99DB70FAE1D7CE227FB6488205B555B38483C65D 3A79BD29 A4A9406876FCBD3C456770C88C718D3B5072E1F5 94E279EB8D8F25B21810ADF121EA45AB2F86D6A1 ) \
+ && gpg --export --armor 430BDF5C56E7C94E848EE60C1C4CBDCDCD2EFD2A > ${GNUPGHOME}/RPM-GPG-KEY-Percona.1 \
+ && gpg --export --armor 4D1BB29D63D98E422B2113B19334A25F8507EFA5 > ${GNUPGHOME}/RPM-GPG-KEY-Percona.2 \
+ && gpg --export --armor 99DB70FAE1D7CE227FB6488205B555B38483C65D > ${GNUPGHOME}/RPM-GPG-KEY-CentOS-8 \
+ && gpg --export --armor 94E279EB8D8F25B21810ADF121EA45AB2F86D6A1 > ${GNUPGHOME}/RPM-GPG-KEY-EPEL-8 \
+ && gpg --export --armor 3A79BD29 > ${GNUPGHOME}/RPM-GPG-KEY-MySQL.1 \
+ && gpg --export --armor A4A9406876FCBD3C456770C88C718D3B5072E1F5 > ${GNUPGHOME}/RPM-GPG-KEY-MySQL.2 \
+ && rpmkeys --import ${GNUPGHOME}/RPM-GPG-KEY-Percona.1 ${GNUPGHOME}/RPM-GPG-KEY-Percona.2 ${GNUPGHOME}/RPM-GPG-KEY-CentOS-8 ${GNUPGHOME}/RPM-GPG-KEY-MySQL.1 ${GNUPGHOME}/RPM-GPG-KEY-MySQL.2 /etc/pki/rpm-gpg/RPM-GPG-KEY-redhat-release ${GNUPGHOME}/RPM-GPG-KEY-EPEL-8 \
+ && curl -L --retry-delay 10 --retry 3 -o /tmp/mysqlrepo.rpm https://dev.mysql.com/get/mysql80-community-release-el8-1.noarch.rpm \
+ && curl -L --retry-delay 10 --retry 3 -o /tmp/perconarepo.rpm https://repo.percona.com/yum/percona-release-latest.noarch.rpm \
+ && rpmkeys --checksig /tmp/mysqlrepo.rpm /tmp/perconarepo.rpm \
+ && rpm -Uvh /tmp/mysqlrepo.rpm /tmp/perconarepo.rpm \
+ && rm -f /tmp/mysqlrepo.rpm /tmp/perconarepo.rpm
+RUN echo H4sICIDAHmICA2ZvbwDVkDFLxEAQhfv9FVfY7o4RhCBsoXJcIXKHwUIOi7m5MVk2yS6zG0//vYlRULTU4rrHvOHN+2ZL5Q4TP6oeO7bX3Od1pcuFXlyNUzVZg7S2yTmmCwDsgzjuDSUyB5SDI2+QzOChcyJBEnwkPOPQZijNuTkrigKmsHUFJ1MeCjUQEqg61tQweVtM0vOrfXItj1eAM0H0DiR2erTgbnOrV5uVvlk+6M+Kinvctby3p0ptqRziHjOnnxz3s/FnKJcxVlkYu/+k4Zcs+AvM8n3+jWW8MBc2NO6FZILUMEsoYQ76UvWI/vAGB/SOZZsCAAA= | base64 -d | gzip -dc > /etc/yum.repos.d/CentOS-Base.repo \
+ && yum install -y --setopt=alwaysprompt=no --setopt=tsflags=nodocs --enablerepo c8base --enablerepo c8updates --enablerepo c8extras libev numactl-libs sysstat strace \
+ && yum install -y --setopt=alwaysprompt=no --setopt=tsflags=nodocs https://download-ib01.fedoraproject.org/pub/epel/8/Everything/x86_64/Packages/g/gperftools-libs-2.7-9.el8.x86_64.rpm https://download-ib01.fedoraproject.org/pub/epel/8/Everything/x86_64/Packages/j/jemalloc-5.2.1-2.el8.x86_64.rpm https://download-ib01.fedoraproject.org/pub/epel/8/Everything/x86_64/Packages/l/libunwind-1.3.1-3.el8.x86_64.rpm
+RUN yum update -y --setopt=alwaysprompt=no --setopt=tsflags=nodocs \
+ && yum install -y --setopt=alwaysprompt=no --setopt=tsflags=nodocs bzip2 ca-certificates gnupg libaio libcurl \
+    procps-ng rsync wget openssl hostname curl tzdata make \
+# Can't use alwaysprompt=no here, since we need to pick up deps
+#   No way to separate key imports and accept deps separately in yum/dnf
+ && yum install -y --setopt=tsflags=nodocs --enablerepo mysql80-community \
+    mysql-community-client mysql-community-server \
+# Have to use hacks to ignore conflicts on /etc/my.cnf install
+ && mkdir -p /tmp/1 \
+ && /usr/bin/percona-release enable-only pxb-80 \
+ && /usr/bin/percona-release enable pt \
+ && yum install -y --setopt=alwaysprompt=no --downloadonly --downloaddir=/tmp/1 --enablerepo mysql80-community percona-xtrabackup-80 percona-toolkit \
+ && rpm -Uvh --replacefiles /tmp/1/*rpm \
+ && rm -rf /tmp/1 \
+ && yum clean all \
+ && yum clean all --enablerepo mysql80-community \
+ && rm -rf /etc/my.cnf /var/lib/mysql /tmp/gpg /sbin/mysqld-debug
+
+# Set up Vitess user and directory tree.
+RUN groupadd -g 1001 -r vitess && useradd -r -u 1001 -g vitess vitess
+RUN mkdir -p /vt/vtdataroot && chown -R vitess:vitess /vt
+
+# Set up Vitess environment (just enough to run pre-built Go binaries)
+ENV VTROOT /vt/src/vitess.io/vitess
+ENV VTDATAROOT /vt/vtdataroot
+ENV PATH $VTROOT/bin:$PATH
+ENV MYSQL_FLAVOR MySQL80
+
+# Copy artifacts from builder layer.
+COPY --from=builder --chown=vitess:vitess /vt/install /vt
+COPY --from=builder --chown=vitess:vitess /vt/src/vitess.io/vitess/web/orchestrator /vt/web/orchestrator
+
+RUN mkdir -p /licenses
+COPY LICENSE /licenses
+
+# Create mount point for actual data (e.g. MySQL data dir)
+VOLUME /vt/vtdataroot
+USER vitess
+
+LABEL name="Vitess Lite image - MySQL Community Server 8.0" \
+      io.k8s.display-name="Vitess Lite image - MySQL Community Server 8.0" \
+      maintainer="cncf-vitess-maintainers@lists.cncf.io" \
+      vendor="CNCF" \
+      version="13.0.0" \
+      release="1" \
+      summary="Vitess base container image, containing Vitess components along with MySQL Community Server 8.0" \
+      description="Vitess base container image, containing Vitess components along with MySQL Community Server 8.0" \
+      io.k8s.description="Vitess base container image, containing Vitess components along with MySQL Community Server 8.0" \
+      distribution-scope="public" \
+      url="https://vitess.io"
diff --git a/go/vt/vtadmin/vtctldclient/proxy.go b/go/vt/vtadmin/vtctldclient/proxy.go
index 8dc90be7da..124d7245ca 100644
--- a/go/vt/vtadmin/vtctldclient/proxy.go
+++ b/go/vt/vtadmin/vtctldclient/proxy.go
@@ -125,7 +125,7 @@ func (vtctld *ClientProxy) Dial(ctx context.Context) error {
 		span.Annotate("is_stale", true)
 
 		// close before reopen. this is safe to call on an already-closed client.
-		if err := vtctld.Close(); err != nil {
+		if err := vtctld.closeLocked(); err != nil {
 			// Even if the client connection does not shut down cleanly, we don't want to block
 			// Dial from discovering a new vtctld. This makes VTAdmin's dialer more resilient,
 			// but, as a caveat, it _can_ potentially leak improperly-closed gRPC connections.
@@ -180,6 +180,13 @@ func (vtctld *ClientProxy) Dial(ctx context.Context) error {
 
 // Close is part of the Proxy interface.
 func (vtctld *ClientProxy) Close() error {
+	vtctld.m.Lock()
+	defer vtctld.m.Unlock()
+
+	return vtctld.closeLocked()
+}
+
+func (vtctld *ClientProxy) closeLocked() error {
 	if vtctld.VtctldClient == nil {
 		vtctld.closed = true
 
diff --git a/go/vt/vtadmin/vtsql/vtsql.go b/go/vt/vtadmin/vtsql/vtsql.go
index 5fcf50fa27..248e10927a 100644
--- a/go/vt/vtadmin/vtsql/vtsql.go
+++ b/go/vt/vtadmin/vtsql/vtsql.go
@@ -21,6 +21,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"google.golang.org/grpc"
@@ -75,6 +76,7 @@ type VTGateProxy struct {
 	DialFunc        func(cfg vitessdriver.Configuration) (*sql.DB, error)
 	dialPingTimeout time.Duration
 
+	m        sync.Mutex
 	host     string
 	conn     *sql.DB
 	dialedAt time.Time
@@ -137,6 +139,9 @@ func (vtgate *VTGateProxy) Dial(ctx context.Context, target string, opts ...grpc
 
 	vtgate.annotateSpan(span)
 
+	vtgate.m.Lock()
+	defer vtgate.m.Unlock()
+
 	if vtgate.conn != nil {
 		ctx, cancel := context.WithTimeout(ctx, vtgate.dialPingTimeout)
 		defer cancel()
@@ -153,7 +158,7 @@ func (vtgate *VTGateProxy) Dial(ctx context.Context, target string, opts ...grpc
 		default:
 			log.Warningf("Ping failed on host %s: %s; Rediscovering a vtgate to get new connection", vtgate.host, err)
 
-			if err := vtgate.Close(); err != nil {
+			if err := vtgate.closeLocked(); err != nil {
 				log.Warningf("Error when closing connection to vtgate %s: %s; Continuing anyway ...", vtgate.host, err)
 			}
 		}
@@ -237,6 +242,13 @@ func (vtgate *VTGateProxy) pingContext(ctx context.Context) error {
 
 // Close is part of the DB interface and satisfies io.Closer.
 func (vtgate *VTGateProxy) Close() error {
+	vtgate.m.Lock()
+	defer vtgate.m.Unlock()
+
+	return vtgate.closeLocked()
+}
+
+func (vtgate *VTGateProxy) closeLocked() error {
 	if vtgate.conn == nil {
 		return nil
 	}
@@ -251,6 +263,9 @@ func (vtgate *VTGateProxy) Close() error {
 
 // Debug implements debug.Debuggable for VTGateProxy.
 func (vtgate *VTGateProxy) Debug() map[string]any {
+	vtgate.m.Lock()
+	defer vtgate.m.Unlock()
+
 	m := map[string]any{
 		"host":         vtgate.host,
 		"is_connected": (vtgate.conn != nil),
diff --git a/go/vt/vtgate/engine/cached_size.go b/go/vt/vtgate/engine/cached_size.go
index 51fc58bb76..58cda04172 100644
--- a/go/vt/vtgate/engine/cached_size.go
+++ b/go/vt/vtgate/engine/cached_size.go
@@ -59,7 +59,7 @@ func (cached *AlterVSchema) CachedSize(alloc bool) int64 {
 	size += cached.AlterVschemaDDL.CachedSize(true)
 	return size
 }
-func (cached *Concatenate) CachedSize(alloc bool) int64 {
+func (cached *CheckCol) CachedSize(alloc bool) int64 {
 	if cached == nil {
 		return int64(0)
 	}
@@ -67,6 +67,20 @@ func (cached *Concatenate) CachedSize(alloc bool) int64 {
 	if alloc {
 		size += int64(24)
 	}
+	// field WsCol *int
+	size += hack.RuntimeAllocSize(int64(8))
+	return size
+}
+
+//go:nocheckptr
+func (cached *Concatenate) CachedSize(alloc bool) int64 {
+	if cached == nil {
+		return int64(0)
+	}
+	size := int64(0)
+	if alloc {
+		size += int64(32)
+	}
 	// field Sources []vitess.io/vitess/go/vt/vtgate/engine.Primitive
 	{
 		size += hack.RuntimeAllocSize(int64(cap(cached.Sources)) * int64(16))
@@ -76,6 +90,17 @@ func (cached *Concatenate) CachedSize(alloc bool) int64 {
 			}
 		}
 	}
+	// field NoNeedToTypeCheck map[int]any
+	if cached.NoNeedToTypeCheck != nil {
+		size += int64(48)
+		hmap := reflect.ValueOf(cached.NoNeedToTypeCheck)
+		numBuckets := int(math.Pow(2, float64((*(*uint8)(unsafe.Pointer(hmap.Pointer() + uintptr(9)))))))
+		numOldBuckets := (*(*uint16)(unsafe.Pointer(hmap.Pointer() + uintptr(10))))
+		size += hack.RuntimeAllocSize(int64(numOldBuckets * 208))
+		if len(cached.NoNeedToTypeCheck) > 0 || numBuckets > 1 {
+			size += hack.RuntimeAllocSize(int64(numBuckets * 208))
+		}
+	}
 	return size
 }
 func (cached *DBDDL) CachedSize(alloc bool) int64 {
@@ -158,9 +183,26 @@ func (cached *Distinct) CachedSize(alloc bool) int64 {
 	if cc, ok := cached.Source.(cachedObject); ok {
 		size += cc.CachedSize(true)
 	}
-	// field ColCollations []vitess.io/vitess/go/mysql/collations.ID
+	// field CheckCols []vitess.io/vitess/go/vt/vtgate/engine.CheckCol
 	{
-		size += hack.RuntimeAllocSize(int64(cap(cached.ColCollations)) * int64(2))
+		size += hack.RuntimeAllocSize(int64(cap(cached.CheckCols)) * int64(18))
+		for _, elem := range cached.CheckCols {
+			size += elem.CachedSize(false)
+		}
+	}
+	return size
+}
+func (cached *DistinctV3) CachedSize(alloc bool) int64 {
+	if cached == nil {
+		return int64(0)
+	}
+	size := int64(0)
+	if alloc {
+		size += int64(16)
+	}
+	// field Source vitess.io/vitess/go/vt/vtgate/engine.Primitive
+	if cc, ok := cached.Source.(cachedObject); ok {
+		size += cc.CachedSize(true)
 	}
 	return size
 }
diff --git a/go/vt/vtgate/engine/concatenate.go b/go/vt/vtgate/engine/concatenate.go
index 1e8eb5b3f1..a2f8dcc7ae 100644
--- a/go/vt/vtgate/engine/concatenate.go
+++ b/go/vt/vtgate/engine/concatenate.go
@@ -31,6 +31,24 @@ var _ Primitive = (*Concatenate)(nil)
 // Concatenate specified the parameter for concatenate primitive
 type Concatenate struct {
 	Sources []Primitive
+
+	// These column offsets do not need to be typed checked - they usually contain weight_string()
+	// columns that are not going to be returned to the user
+	NoNeedToTypeCheck map[int]any
+}
+
+// NewConcatenate creates a Concatenate primitive. The ignoreCols slice contains the offsets that
+// don't need to have the same type between sources -
+// weight_string() sometimes returns VARBINARY and sometimes VARCHAR
+func NewConcatenate(Sources []Primitive, ignoreCols []int) *Concatenate {
+	ignoreTypes := map[int]any{}
+	for _, i := range ignoreCols {
+		ignoreTypes[i] = nil
+	}
+	return &Concatenate{
+		Sources:           Sources,
+		NoNeedToTypeCheck: ignoreTypes,
+	}
 }
 
 // RouteType returns a description of the query routing type used by the primitive
@@ -101,23 +119,28 @@ func (c *Concatenate) TryExecute(vcursor VCursor, bindVars map[string]*querypb.B
 }
 
 func (c *Concatenate) getFields(res []*sqltypes.Result) ([]*querypb.Field, error) {
-	var resFields []*querypb.Field
+	if len(res) == 0 {
+		return nil, nil
+	}
+
+	var fields []*querypb.Field
 	for _, r := range res {
-		fields := r.Fields
-		if fields == nil {
+		if r.Fields == nil {
 			continue
 		}
-		if resFields == nil {
-			resFields = fields
+		if fields == nil {
+			fields = r.Fields
 			continue
 		}
-		err := compareFields(fields, resFields)
+
+		err := c.compareFields(fields, r.Fields)
 		if err != nil {
 			return nil, err
 		}
 	}
-	return resFields, nil
+	return fields, nil
 }
+
 func (c *Concatenate) execSources(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool) ([]*sqltypes.Result, error) {
 	results := make([]*sqltypes.Result, len(c.Sources))
 	g, restoreCtx := vcursor.ErrorGroupCancellableContext()
@@ -172,7 +195,7 @@ func (c *Concatenate) TryStreamExecute(vcursor VCursor, bindVars map[string]*que
 				}
 				wg.Wait()
 				if resultChunk.Fields != nil {
-					err := compareFields(seenFields, resultChunk.Fields)
+					err := c.compareFields(seenFields, resultChunk.Fields)
 					if err != nil {
 						return err
 					}
@@ -209,6 +232,7 @@ func (c *Concatenate) TryStreamExecute(vcursor VCursor, bindVars map[string]*que
 
 // GetFields fetches the field info.
 func (c *Concatenate) GetFields(vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
+	// TODO: type coercions
 	res, err := c.Sources[0].GetFields(vcursor, bindVars)
 	if err != nil {
 		return nil, err
@@ -219,11 +243,12 @@ func (c *Concatenate) GetFields(vcursor VCursor, bindVars map[string]*querypb.Bi
 		if err != nil {
 			return nil, err
 		}
-		err = compareFields(result.Fields, res.Fields)
+		err = c.compareFields(res.Fields, result.Fields)
 		if err != nil {
 			return nil, err
 		}
 	}
+
 	return res, nil
 }
 
@@ -246,12 +271,15 @@ func (c *Concatenate) description() PrimitiveDescription {
 	return PrimitiveDescription{OperatorType: c.RouteType()}
 }
 
-func compareFields(fields1 []*querypb.Field, fields2 []*querypb.Field) error {
+func (c *Concatenate) compareFields(fields1 []*querypb.Field, fields2 []*querypb.Field) error {
 	if len(fields1) != len(fields2) {
 		return ErrWrongNumberOfColumnsInSelect
 	}
-	for i, field2 := range fields2 {
-		field1 := fields1[i]
+	for i, field1 := range fields1 {
+		if _, found := c.NoNeedToTypeCheck[i]; found {
+			continue
+		}
+		field2 := fields2[i]
 		if field1.Type != field2.Type {
 			return vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "merging field of different types is not supported, name: (%v, %v) types: (%v, %v)", field1.Name, field2.Name, field1.Type, field2.Type)
 		}
diff --git a/go/vt/vtgate/engine/distinct.go b/go/vt/vtgate/engine/distinct.go
index 4ed9f34384..026b359eb8 100644
--- a/go/vt/vtgate/engine/distinct.go
+++ b/go/vt/vtgate/engine/distinct.go
@@ -17,32 +17,40 @@ limitations under the License.
 package engine
 
 import (
+	"fmt"
+
 	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/sqltypes"
 	querypb "vitess.io/vitess/go/vt/proto/query"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
 )
 
 // Distinct Primitive is used to uniqueify results
 var _ Primitive = (*Distinct)(nil)
 
-// Distinct Primitive is used to uniqueify results
-type Distinct struct {
-	Source        Primitive
-	ColCollations []collations.ID
-}
-
-type row = []sqltypes.Value
-
-type probeTable struct {
-	seenRows      map[evalengine.HashCode][]row
-	colCollations []collations.ID
-}
+type (
+	// Distinct Primitive is used to uniqueify results
+	Distinct struct {
+		Source    Primitive
+		CheckCols []CheckCol
+		Truncate  bool
+	}
+	CheckCol struct {
+		Col       int
+		WsCol     *int
+		Collation collations.ID
+	}
+	probeTable struct {
+		seenRows  map[evalengine.HashCode][]sqltypes.Row
+		checkCols []CheckCol
+	}
+)
 
-func (pt *probeTable) exists(inputRow row) (bool, error) {
+func (pt *probeTable) exists(inputRow sqltypes.Row) (bool, error) {
 	// the two prime numbers used here (17 and 31) are used to
-
-	// calculate hashcode from all column values in the input row
+	// calculate hashcode from all column values in the input sqltypes.Row
 	code, err := pt.hashCodeForRow(inputRow)
 	if err != nil {
 		return false, err
@@ -50,15 +58,15 @@ func (pt *probeTable) exists(inputRow row) (bool, error) {
 
 	existingRows, found := pt.seenRows[code]
 	if !found {
-		// nothing with this hash code found, we can be sure it's a not seen row
-		pt.seenRows[code] = []row{inputRow}
+		// nothing with this hash code found, we can be sure it's a not seen sqltypes.Row
+		pt.seenRows[code] = []sqltypes.Row{inputRow}
 		return false, nil
 	}
 
 	// we found something in the map - still need to check all individual values
 	// so we don't just fall for a hash collision
 	for _, existingRow := range existingRows {
-		exists, err := equal(existingRow, inputRow, pt.colCollations)
+		exists, err := pt.equal(existingRow, inputRow)
 		if err != nil {
 			return false, err
 		}
@@ -72,7 +80,7 @@ func (pt *probeTable) exists(inputRow row) (bool, error) {
 	return false, nil
 }
 
-func (pt *probeTable) hashCodeForRow(inputRow row) (evalengine.HashCode, error) {
+func (pt *probeTable) hashCodeForRow(inputRow sqltypes.Row) (evalengine.HashCode, error) {
 	// Why use 17 and 31 in this method?
 	// Copied from an old usenet discussion on the topic:
 	// https://groups.google.com/g/comp.programming/c/HSurZEyrZ1E?pli=1#d887b5bdb2dac99d
@@ -105,32 +113,42 @@ func (pt *probeTable) hashCodeForRow(inputRow row) (evalengine.HashCode, error)
 	// > So you want an odd MULT that has plenty of one-bits.
 
 	code := evalengine.HashCode(17)
-	for idx, value := range inputRow {
-		// We use unknown collations when we do not have collation information
-		// This is safe for types which do not require collation information like
-		// numeric types. It will fail at runtime for text types.
-		collation := collations.Unknown
-		if len(pt.colCollations) > idx {
-			collation = pt.colCollations[idx]
+	for i, checkCol := range pt.checkCols {
+		if i >= len(inputRow) {
+			return 0, vterrors.New(vtrpcpb.Code_INTERNAL, "distinct check colls is larger than its input row")
 		}
-		hashcode, err := evalengine.NullsafeHashcode(value, collation, value.Type())
+		col := inputRow[checkCol.Col]
+		hashcode, err := evalengine.NullsafeHashcode(col, checkCol.Collation, col.Type())
 		if err != nil {
-			return 0, err
+			if err != evalengine.UnsupportedCollationHashError || checkCol.WsCol == nil {
+				return 0, err
+			}
+			checkCol = checkCol.SwitchToWeightString()
+			pt.checkCols[i] = checkCol
+			hashcode, err = evalengine.NullsafeHashcode(inputRow[checkCol.Col], checkCol.Collation, col.Type())
+			if err != nil {
+				return 0, err
+			}
 		}
 		code = code*31 + hashcode
 	}
 	return code, nil
 }
 
-func equal(a, b []sqltypes.Value, colCollations []collations.ID) (bool, error) {
-	for i, aVal := range a {
-		collation := collations.Unknown
-		if len(colCollations) > i {
-			collation = colCollations[i]
-		}
-		cmp, err := evalengine.NullsafeCompare(aVal, b[i], collation)
+func (pt *probeTable) equal(a, b sqltypes.Row) (bool, error) {
+	for i, checkCol := range pt.checkCols {
+		cmp, err := evalengine.NullsafeCompare(a[i], b[i], checkCol.Collation)
 		if err != nil {
-			return false, err
+			_, isComparisonErr := err.(evalengine.UnsupportedComparisonError)
+			if !isComparisonErr || checkCol.WsCol == nil {
+				return false, err
+			}
+			checkCol = checkCol.SwitchToWeightString()
+			pt.checkCols[i] = checkCol
+			cmp, err = evalengine.NullsafeCompare(a[i], b[i], checkCol.Collation)
+			if err != nil {
+				return false, err
+			}
 		}
 		if cmp != 0 {
 			return false, nil
@@ -139,10 +157,12 @@ func equal(a, b []sqltypes.Value, colCollations []collations.ID) (bool, error) {
 	return true, nil
 }
 
-func newProbeTable(colCollations []collations.ID) *probeTable {
+func newProbeTable(checkCols []CheckCol) *probeTable {
+	cols := make([]CheckCol, len(checkCols))
+	copy(cols, checkCols)
 	return &probeTable{
-		seenRows:      map[uintptr][]row{},
-		colCollations: colCollations,
+		seenRows:  map[uintptr][]sqltypes.Row{},
+		checkCols: cols,
 	}
 }
 
@@ -158,7 +178,7 @@ func (d *Distinct) TryExecute(vcursor VCursor, bindVars map[string]*querypb.Bind
 		InsertID: input.InsertID,
 	}
 
-	pt := newProbeTable(d.ColCollations)
+	pt := newProbeTable(d.CheckCols)
 
 	for _, row := range input.Rows {
 		exists, err := pt.exists(row)
@@ -169,13 +189,15 @@ func (d *Distinct) TryExecute(vcursor VCursor, bindVars map[string]*querypb.Bind
 			result.Rows = append(result.Rows, row)
 		}
 	}
-
+	if d.Truncate {
+		return result.Truncate(len(d.CheckCols)), nil
+	}
 	return result, err
 }
 
 // TryStreamExecute implements the Primitive interface
 func (d *Distinct) TryStreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
-	pt := newProbeTable(d.ColCollations)
+	pt := newProbeTable(d.CheckCols)
 
 	err := vcursor.StreamExecutePrimitive(d.Source, bindVars, wantfields, func(input *sqltypes.Result) error {
 		result := &sqltypes.Result{
@@ -191,7 +213,7 @@ func (d *Distinct) TryStreamExecute(vcursor VCursor, bindVars map[string]*queryp
 				result.Rows = append(result.Rows, row)
 			}
 		}
-		return callback(result)
+		return callback(result.Truncate(len(d.CheckCols)))
 	})
 
 	return err
@@ -228,26 +250,46 @@ func (d *Distinct) Inputs() []Primitive {
 }
 
 func (d *Distinct) description() PrimitiveDescription {
-	var other map[string]any
-	if d.ColCollations != nil {
-		allUnknown := true
-		other = map[string]any{}
-		var colls []string
-		for _, collation := range d.ColCollations {
-			coll := collations.Local().LookupByID(collation)
-			if coll == nil {
-				colls = append(colls, "UNKNOWN")
-			} else {
-				colls = append(colls, coll.Name())
-				allUnknown = false
-			}
-		}
-		if !allUnknown {
-			other["Collations"] = colls
-		}
+	other := map[string]any{}
+
+	var colls []string
+	for _, checkCol := range d.CheckCols {
+		colls = append(colls, checkCol.String())
+	}
+	if colls != nil {
+		other["Collations"] = colls
+	}
+
+	if d.Truncate {
+		other["ResultColumns"] = len(d.CheckCols)
 	}
 	return PrimitiveDescription{
 		Other:        other,
 		OperatorType: "Distinct",
 	}
 }
+
+// SwitchToWeightString returns a new CheckCol that works on the weight string column instead
+func (cc CheckCol) SwitchToWeightString() CheckCol {
+	return CheckCol{
+		Col:       *cc.WsCol,
+		WsCol:     nil,
+		Collation: collations.CollationBinaryID,
+	}
+}
+
+func (cc CheckCol) String() string {
+	coll := collations.Local().LookupByID(cc.Collation)
+	var collation string
+	if coll != nil {
+		collation = ": " + coll.Name()
+	}
+
+	var column string
+	if cc.WsCol == nil {
+		column = fmt.Sprintf("%d", cc.Col)
+	} else {
+		column = fmt.Sprintf("(%d:%d)", cc.Col, *cc.WsCol)
+	}
+	return column + collation
+}
diff --git a/go/vt/vtgate/engine/distinctV3.go b/go/vt/vtgate/engine/distinctV3.go
new file mode 100644
index 0000000000..9992c4def7
--- /dev/null
+++ b/go/vt/vtgate/engine/distinctV3.go
@@ -0,0 +1,178 @@
+/*
+Copyright 2020 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package engine
+
+import (
+	"vitess.io/vitess/go/mysql/collations"
+	"vitess.io/vitess/go/sqltypes"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+	"vitess.io/vitess/go/vt/vtgate/evalengine"
+)
+
+// DistinctV3 Primitive is used to uniqueify results
+// It does not always work, and should be removed once the V3 planner has been removed
+var _ Primitive = (*DistinctV3)(nil)
+
+// Distinct Primitive is used to uniqueify results
+type DistinctV3 struct {
+	Source Primitive
+}
+
+type row = []sqltypes.Value
+
+type probeTableV3 struct {
+	m map[evalengine.HashCode][]row
+}
+
+func (pt *probeTableV3) exists(inputRow row) (bool, error) {
+	// calculate hashcode from all column values in the input row
+	code := evalengine.HashCode(17)
+	for _, value := range inputRow {
+		hashcode, err := evalengine.NullsafeHashcode(value, collations.Unknown, value.Type())
+		if err != nil {
+			return false, err
+		}
+		code = code*31 + hashcode
+	}
+
+	existingRows, found := pt.m[code]
+	if !found {
+		// nothing with this hash code found, we can be sure it's a not seen row
+		pt.m[code] = []row{inputRow}
+		return false, nil
+	}
+
+	// we found something in the map - still need to check all individual values
+	// so we don't just fall for a hash collision
+	for _, existingRow := range existingRows {
+		exists, err := equalV3(existingRow, inputRow)
+		if err != nil {
+			return false, err
+		}
+		if exists {
+			return true, nil
+		}
+	}
+
+	pt.m[code] = append(existingRows, inputRow)
+
+	return false, nil
+}
+
+func equalV3(a, b []sqltypes.Value) (bool, error) {
+	for i, aVal := range a {
+		cmp, err := evalengine.NullsafeCompare(aVal, b[i], collations.Unknown)
+		if err != nil {
+			return false, err
+		}
+		if cmp != 0 {
+			return false, nil
+		}
+	}
+	return true, nil
+}
+
+func newProbeTableV3() *probeTableV3 {
+	return &probeTableV3{m: map[evalengine.HashCode][]row{}}
+}
+
+// TryExecute implements the Primitive interface
+func (d *DistinctV3) TryExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool) (*sqltypes.Result, error) {
+	input, err := vcursor.ExecutePrimitive(d.Source, bindVars, wantfields)
+	if err != nil {
+		return nil, err
+	}
+
+	result := &sqltypes.Result{
+		Fields:   input.Fields,
+		InsertID: input.InsertID,
+	}
+
+	pt := newProbeTableV3()
+
+	for _, row := range input.Rows {
+		exists, err := pt.exists(row)
+		if err != nil {
+			return nil, err
+		}
+		if !exists {
+			result.Rows = append(result.Rows, row)
+		}
+	}
+
+	return result, err
+}
+
+// TryStreamExecute implements the Primitive interface
+func (d *DistinctV3) TryStreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
+	pt := newProbeTableV3()
+
+	err := vcursor.StreamExecutePrimitive(d.Source, bindVars, wantfields, func(input *sqltypes.Result) error {
+		result := &sqltypes.Result{
+			Fields:   input.Fields,
+			InsertID: input.InsertID,
+		}
+		for _, row := range input.Rows {
+			exists, err := pt.exists(row)
+			if err != nil {
+				return err
+			}
+			if !exists {
+				result.Rows = append(result.Rows, row)
+			}
+		}
+		return callback(result)
+	})
+
+	return err
+}
+
+// RouteType implements the Primitive interface
+func (d *DistinctV3) RouteType() string {
+	return d.Source.RouteType()
+}
+
+// GetKeyspaceName implements the Primitive interface
+func (d *DistinctV3) GetKeyspaceName() string {
+	return d.Source.GetKeyspaceName()
+}
+
+// GetTableName implements the Primitive interface
+func (d *DistinctV3) GetTableName() string {
+	return d.Source.GetTableName()
+}
+
+// GetFields implements the Primitive interface
+func (d *DistinctV3) GetFields(vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
+	return d.Source.GetFields(vcursor, bindVars)
+}
+
+// NeedsTransaction implements the Primitive interface
+func (d *DistinctV3) NeedsTransaction() bool {
+	return d.Source.NeedsTransaction()
+}
+
+// Inputs implements the Primitive interface
+func (d *DistinctV3) Inputs() []Primitive {
+	return []Primitive{d.Source}
+}
+
+func (d *DistinctV3) description() PrimitiveDescription {
+	return PrimitiveDescription{
+		OperatorType: "Distinct",
+	}
+}
diff --git a/go/vt/vtgate/engine/hash_join.go b/go/vt/vtgate/engine/hash_join.go
index b97f7a995a..3e06a53625 100644
--- a/go/vt/vtgate/engine/hash_join.go
+++ b/go/vt/vtgate/engine/hash_join.go
@@ -112,8 +112,8 @@ func (hj *HashJoin) TryExecute(vcursor VCursor, bindVars map[string]*querypb.Bin
 	return result, nil
 }
 
-func (hj *HashJoin) buildProbeTable(lresult *sqltypes.Result) (map[evalengine.HashCode][]row, error) {
-	probeTable := map[evalengine.HashCode][]row{}
+func (hj *HashJoin) buildProbeTable(lresult *sqltypes.Result) (map[evalengine.HashCode][]sqltypes.Row, error) {
+	probeTable := map[evalengine.HashCode][]sqltypes.Row{}
 	for _, current := range lresult.Rows {
 		joinVal := current[hj.LHSKey]
 		if joinVal.IsNull() {
@@ -131,7 +131,7 @@ func (hj *HashJoin) buildProbeTable(lresult *sqltypes.Result) (map[evalengine.Ha
 // TryStreamExecute implements the Primitive interface
 func (hj *HashJoin) TryStreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
 	// build the probe table from the LHS result
-	probeTable := map[evalengine.HashCode][]row{}
+	probeTable := map[evalengine.HashCode][]sqltypes.Row{}
 	var lfields []*querypb.Field
 	err := vcursor.StreamExecutePrimitive(hj.Left, bindVars, wantfields, func(result *sqltypes.Result) error {
 		if len(lfields) == 0 && len(result.Fields) != 0 {
diff --git a/go/vt/vtgate/evalengine/comparisons.go b/go/vt/vtgate/evalengine/comparisons.go
index aa3510d9bb..5372dfc1b2 100644
--- a/go/vt/vtgate/evalengine/comparisons.go
+++ b/go/vt/vtgate/evalengine/comparisons.go
@@ -146,7 +146,7 @@ func evalResultsAreDateAndNumeric(l, r *EvalResult) bool {
 func compareAsTuples(lVal, rVal *EvalResult) bool {
 	left := lVal.typeof() == sqltypes.Tuple
 	right := rVal.typeof() == sqltypes.Tuple
-	if left && right {
+	if right && left {
 		return true
 	}
 	if left || right {
@@ -408,6 +408,9 @@ func (err UnsupportedCollationError) Error() string {
 	return fmt.Sprintf("cannot compare strings, collation is unknown or unsupported (collation ID: %d)", err.ID)
 }
 
+// UnsupportedCollationHashError is returned when we try to get the hash value and are missing the collation to use
+var UnsupportedCollationHashError = vterrors.Errorf(vtrpcpb.Code_INTERNAL, "text type with an unknown/unsupported collation cannot be hashed")
+
 // NullsafeCompare returns 0 if v1==v2, -1 if v1<v2, and 1 if v1>v2.
 // NULL is the lowest value. If any value is
 // numeric, then a numeric comparison is performed after
diff --git a/go/vt/vtgate/evalengine/eval_result.go b/go/vt/vtgate/evalengine/eval_result.go
index 4361e2aa5a..2e7c5d29ee 100644
--- a/go/vt/vtgate/evalengine/eval_result.go
+++ b/go/vt/vtgate/evalengine/eval_result.go
@@ -465,7 +465,7 @@ func (er *EvalResult) nullSafeHashcode() (HashCode, error) {
 	case er.isTextual():
 		coll := collations.Local().LookupByID(er.collation().Collation)
 		if coll == nil {
-			return 0, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "text type with an unknown/unsupported collation cannot be hashed")
+			return 0, UnsupportedCollationHashError
 		}
 		return coll.Hash(er.bytes(), 0), nil
 	case sqltypes.IsDate(er.typeof()):
diff --git a/go/vt/vtgate/planbuilder/builder.go b/go/vt/vtgate/planbuilder/builder.go
index 7dbf06e9d0..f97ac6b7ef 100644
--- a/go/vt/vtgate/planbuilder/builder.go
+++ b/go/vt/vtgate/planbuilder/builder.go
@@ -393,19 +393,16 @@ func buildFlushTables(stmt *sqlparser.Flush, vschema plancontext.VSchema) (engin
 		return keys[i].ks.Name < keys[j].ks.Name
 	})
 
-	finalPlan := &engine.Concatenate{
-		Sources: nil,
-	}
+	var sources []engine.Primitive
 	for _, sendDest := range keys {
 		plan := &engine.Send{
 			Keyspace:          sendDest.ks,
 			TargetDestination: sendDest.dest,
 			Query:             sqlparser.String(newFlushStmt(stmt, tablesMap[sendDest])),
 		}
-		finalPlan.Sources = append(finalPlan.Sources, plan)
+		sources = append(sources, plan)
 	}
-
-	return finalPlan, nil
+	return engine.NewConcatenate(sources, nil), nil
 }
 
 func newFlushStmt(stmt *sqlparser.Flush, tables sqlparser.TableNames) *sqlparser.Flush {
diff --git a/go/vt/vtgate/planbuilder/concatenate.go b/go/vt/vtgate/planbuilder/concatenate.go
index 8c6d43a8d2..92e31d3de7 100644
--- a/go/vt/vtgate/planbuilder/concatenate.go
+++ b/go/vt/vtgate/planbuilder/concatenate.go
@@ -70,9 +70,7 @@ func (c *concatenate) Primitive() engine.Primitive {
 	lhs := c.lhs.Primitive()
 	rhs := c.rhs.Primitive()
 
-	return &engine.Concatenate{
-		Sources: []engine.Primitive{lhs, rhs},
-	}
+	return engine.NewConcatenate([]engine.Primitive{lhs, rhs}, nil)
 }
 
 // Rewrite implements the logicalPlan interface
diff --git a/go/vt/vtgate/planbuilder/concatenateGen4.go b/go/vt/vtgate/planbuilder/concatenateGen4.go
index 29617b2fb8..8c14b1ba0f 100644
--- a/go/vt/vtgate/planbuilder/concatenateGen4.go
+++ b/go/vt/vtgate/planbuilder/concatenateGen4.go
@@ -26,6 +26,10 @@ import (
 
 type concatenateGen4 struct {
 	sources []logicalPlan
+
+	// These column offsets do not need to be typed checked - they usually contain weight_string()
+	// columns that are not going to be returned to the user
+	noNeedToTypeCheck []int
 }
 
 var _ logicalPlan = (*concatenateGen4)(nil)
@@ -82,9 +86,8 @@ func (c *concatenateGen4) Primitive() engine.Primitive {
 	for _, source := range c.sources {
 		sources = append(sources, source.Primitive())
 	}
-	return &engine.Concatenate{
-		Sources: sources,
-	}
+
+	return engine.NewConcatenate(sources, c.noNeedToTypeCheck)
 }
 
 // Rewrite implements the logicalPlan interface
diff --git a/go/vt/vtgate/planbuilder/distinct.go b/go/vt/vtgate/planbuilder/distinct.go
index 40cb3d6030..695cc43905 100644
--- a/go/vt/vtgate/planbuilder/distinct.go
+++ b/go/vt/vtgate/planbuilder/distinct.go
@@ -17,7 +17,6 @@ limitations under the License.
 package planbuilder
 
 import (
-	"vitess.io/vitess/go/mysql/collations"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vtgate/engine"
@@ -28,20 +27,40 @@ var _ logicalPlan = (*distinct)(nil)
 // distinct is the logicalPlan for engine.Distinct.
 type distinct struct {
 	logicalPlanCommon
-	ColCollations []collations.ID
+	checkCols      []engine.CheckCol
+	needToTruncate bool
 }
 
-func newDistinct(source logicalPlan, colCollations []collations.ID) logicalPlan {
+func newDistinct(source logicalPlan, checkCols []engine.CheckCol, needToTruncate bool) logicalPlan {
 	return &distinct{
 		logicalPlanCommon: newBuilderCommon(source),
-		ColCollations:     colCollations,
+		checkCols:         checkCols,
+		needToTruncate:    needToTruncate,
 	}
 }
 
+func newDistinctV3(source logicalPlan) logicalPlan {
+	return &distinct{logicalPlanCommon: newBuilderCommon(source)}
+}
+
 func (d *distinct) Primitive() engine.Primitive {
+	if d.checkCols == nil {
+		// If we are missing the checkCols information, we are on the V3 planner and should produce a V3 Distinct
+		return &engine.DistinctV3{Source: d.input.Primitive()}
+	}
+	truncate := false
+	if d.needToTruncate {
+		for _, col := range d.checkCols {
+			if col.WsCol != nil {
+				truncate = true
+				break
+			}
+		}
+	}
 	return &engine.Distinct{
-		Source:        d.input.Primitive(),
-		ColCollations: d.ColCollations,
+		Source:    d.input.Primitive(),
+		CheckCols: d.checkCols,
+		Truncate:  truncate,
 	}
 }
 
diff --git a/go/vt/vtgate/planbuilder/gen4_planner.go b/go/vt/vtgate/planbuilder/gen4_planner.go
index 46946e08c8..54808a8a1c 100644
--- a/go/vt/vtgate/planbuilder/gen4_planner.go
+++ b/go/vt/vtgate/planbuilder/gen4_planner.go
@@ -173,12 +173,12 @@ func newBuildSelectPlan(
 		return nil, err
 	}
 
-	plan, err := transformToLogicalPlan(ctx, physOp)
+	plan, err := transformToLogicalPlan(ctx, physOp, true)
 	if err != nil {
 		return nil, err
 	}
 
-	plan, err = planHorizon(ctx, plan, selStmt)
+	plan, err = planHorizon(ctx, plan, selStmt, true)
 	if err != nil {
 		return nil, err
 	}
@@ -225,7 +225,7 @@ func planLimit(limit *sqlparser.Limit, plan logicalPlan) (logicalPlan, error) {
 	return lPlan, nil
 }
 
-func planHorizon(ctx *plancontext.PlanningContext, plan logicalPlan, in sqlparser.SelectStatement) (logicalPlan, error) {
+func planHorizon(ctx *plancontext.PlanningContext, plan logicalPlan, in sqlparser.SelectStatement, truncateColumns bool) (logicalPlan, error) {
 	switch node := in.(type) {
 	case *sqlparser.Select:
 		hp := horizonPlanning{
@@ -234,7 +234,7 @@ func planHorizon(ctx *plancontext.PlanningContext, plan logicalPlan, in sqlparse
 
 		replaceSubQuery(ctx, node)
 		var err error
-		plan, err = hp.planHorizon(ctx, plan)
+		plan, err = hp.planHorizon(ctx, plan, truncateColumns)
 		if err != nil {
 			return nil, err
 		}
diff --git a/go/vt/vtgate/planbuilder/grouping.go b/go/vt/vtgate/planbuilder/grouping.go
index 2bf58dcfa4..d96534f5ce 100644
--- a/go/vt/vtgate/planbuilder/grouping.go
+++ b/go/vt/vtgate/planbuilder/grouping.go
@@ -108,7 +108,7 @@ func planDistinct(input logicalPlan) (logicalPlan, error) {
 			// So, the distinct 'operator' cannot be pushed down into the
 			// route.
 			if rc.column.Origin() == node {
-				return newDistinct(node, nil), nil
+				return newDistinctV3(node), nil
 			}
 			node.groupByKeys = append(node.groupByKeys, &engine.GroupByParams{KeyCol: i, WeightStringCol: -1, FromGroupBy: false})
 		}
diff --git a/go/vt/vtgate/planbuilder/horizon_planning.go b/go/vt/vtgate/planbuilder/horizon_planning.go
index bb0267aadd..401de2ada5 100644
--- a/go/vt/vtgate/planbuilder/horizon_planning.go
+++ b/go/vt/vtgate/planbuilder/horizon_planning.go
@@ -37,7 +37,7 @@ type horizonPlanning struct {
 	qp  *abstract.QueryProjection
 }
 
-func (hp *horizonPlanning) planHorizon(ctx *plancontext.PlanningContext, plan logicalPlan) (logicalPlan, error) {
+func (hp *horizonPlanning) planHorizon(ctx *plancontext.PlanningContext, plan logicalPlan, truncateColumns bool) (logicalPlan, error) {
 	rb, isRoute := plan.(*routeGen4)
 	if !isRoute && ctx.SemTable.NotSingleRouteErr != nil {
 		// If we got here, we don't have a single shard plan
@@ -63,24 +63,23 @@ func (hp *horizonPlanning) planHorizon(ctx *plancontext.PlanningContext, plan lo
 
 	// If we still have a HAVING clause, it's because it could not be pushed to the WHERE,
 	// so it probably has aggregations
-	if hp.qp.NeedsAggregation() || hp.sel.Having != nil {
+	switch {
+	case hp.qp.NeedsAggregation() || hp.sel.Having != nil:
 		plan, err = hp.planAggregations(ctx, plan)
 		if err != nil {
 			return nil, err
 		}
 		// if we already did sorting, we don't need to do it again
 		needsOrdering = needsOrdering && !hp.qp.CanPushDownSorting
-	} else {
-		if canShortcut {
-			err = planSingleShardRoutePlan(hp.sel, rb)
-			if err != nil {
-				return nil, err
-			}
-		} else {
-			err = pushProjections(ctx, plan, hp.qp.SelectExprs)
-			if err != nil {
-				return nil, err
-			}
+	case canShortcut:
+		err = planSingleShardRoutePlan(hp.sel, rb)
+		if err != nil {
+			return nil, err
+		}
+	default:
+		err = pushProjections(ctx, plan, hp.qp.SelectExprs)
+		if err != nil {
+			return nil, err
 		}
 	}
 
@@ -97,6 +96,10 @@ func (hp *horizonPlanning) planHorizon(ctx *plancontext.PlanningContext, plan lo
 		return nil, err
 	}
 
+	if !truncateColumns {
+		return plan, nil
+	}
+
 	plan, err = hp.truncateColumnsIfNeeded(ctx, plan)
 	if err != nil {
 		return nil, err
@@ -1219,7 +1222,7 @@ func (hp *horizonPlanning) addDistinct(ctx *plancontext.PlanningContext, plan lo
 			inner = sqlparser.NewColName(aliasExpr.As.String())
 			ctx.SemTable.CopyDependencies(aliasExpr.Expr, inner)
 		}
-		grpParam := &engine.GroupByParams{KeyCol: index, WeightStringCol: -1, CollationID: ctx.SemTable.CollationForExpr(inner)}
+		grpParam := &engine.GroupByParams{KeyCol: index, WeightStringCol: -1, CollationID: ctx.SemTable.CollationForExpr(inner), Expr: inner}
 		_, wOffset, err := wrapAndPushExpr(ctx, aliasExpr.Expr, aliasExpr.Expr, plan)
 		if err != nil {
 			return nil, err
diff --git a/go/vt/vtgate/planbuilder/operator_transformers.go b/go/vt/vtgate/planbuilder/operator_transformers.go
index 724d65c8af..ef7553bcb9 100644
--- a/go/vt/vtgate/planbuilder/operator_transformers.go
+++ b/go/vt/vtgate/planbuilder/operator_transformers.go
@@ -21,6 +21,8 @@ import (
 	"strconv"
 	"strings"
 
+	"vitess.io/vitess/go/sqltypes"
+
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
 
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/physical"
@@ -39,14 +41,14 @@ import (
 	"vitess.io/vitess/go/vt/vterrors"
 )
 
-func transformToLogicalPlan(ctx *plancontext.PlanningContext, op abstract.PhysicalOperator) (logicalPlan, error) {
+func transformToLogicalPlan(ctx *plancontext.PlanningContext, op abstract.PhysicalOperator, isRoot bool) (logicalPlan, error) {
 	switch op := op.(type) {
 	case *physical.Route:
 		return transformRoutePlan(ctx, op)
 	case *physical.ApplyJoin:
 		return transformApplyJoinPlan(ctx, op)
 	case *physical.Union:
-		return transformUnionPlan(ctx, op)
+		return transformUnionPlan(ctx, op, isRoot)
 	case *physical.Vindex:
 		return transformVindexPlan(ctx, op)
 	case *physical.SubQueryOp:
@@ -56,7 +58,7 @@ func transformToLogicalPlan(ctx *plancontext.PlanningContext, op abstract.Physic
 	case *physical.Derived:
 		return transformDerivedPlan(ctx, op)
 	case *physical.Filter:
-		plan, err := transformToLogicalPlan(ctx, op.Source)
+		plan, err := transformToLogicalPlan(ctx, op.Source, false)
 		if err != nil {
 			return nil, err
 		}
@@ -91,11 +93,11 @@ func transformApplyJoinPlan(ctx *plancontext.PlanningContext, n *physical.ApplyJ
 	//	return nil, err
 	// }
 
-	lhs, err := transformToLogicalPlan(ctx, n.LHS)
+	lhs, err := transformToLogicalPlan(ctx, n.LHS, false)
 	if err != nil {
 		return nil, err
 	}
-	rhs, err := transformToLogicalPlan(ctx, n.RHS)
+	rhs, err := transformToLogicalPlan(ctx, n.RHS, false)
 	if err != nil {
 		return nil, err
 	}
@@ -237,7 +239,7 @@ func getAllTableNames(op *physical.Route) ([]string, error) {
 	return tableNames, nil
 }
 
-func transformUnionPlan(ctx *plancontext.PlanningContext, op *physical.Union) (logicalPlan, error) {
+func transformUnionPlan(ctx *plancontext.PlanningContext, op *physical.Union, isRoot bool) (logicalPlan, error) {
 	var sources []logicalPlan
 	var err error
 	if op.Distinct {
@@ -272,12 +274,98 @@ func transformUnionPlan(ctx *plancontext.PlanningContext, op *physical.Union) (l
 		result = &concatenateGen4{sources: sources}
 	}
 	if op.Distinct {
-		return newDistinct(result, getCollationsFor(ctx, op)), nil
+		colls := getCollationsFor(ctx, op)
+		checkCols, err := getCheckColsForUnion(ctx, result, colls)
+		if err != nil {
+			return nil, err
+		}
+		return newDistinct(result, checkCols, isRoot), nil
 	}
 	return result, nil
 
 }
 
+func getWeightStringForSelectExpr(selectExpr sqlparser.SelectExpr) (*sqlparser.AliasedExpr, error) {
+	expr, isAliased := selectExpr.(*sqlparser.AliasedExpr)
+	if !isAliased {
+		return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "cannot convert select expression to an aliased expression")
+	}
+	return &sqlparser.AliasedExpr{Expr: weightStringFor(expr.Expr)}, nil
+}
+
+func getCheckColsForUnion(ctx *plancontext.PlanningContext, result logicalPlan, colls []collations.ID) ([]engine.CheckCol, error) {
+	checkCols := make([]engine.CheckCol, 0, len(colls))
+	for i, coll := range colls {
+		checkCol := engine.CheckCol{Col: i, Collation: coll}
+		if coll != collations.Unknown {
+			checkCols = append(checkCols, checkCol)
+			continue
+		}
+		// We might need a weight string - let's push one
+		// `might` because we just don't know what type we are dealing with.
+		// If we encounter a numerical value, we don't need any weight_string values
+		newOffset, err := pushWeightStringForDistinct(ctx, result, i)
+		if err != nil {
+			return nil, err
+		}
+		checkCol.WsCol = &newOffset
+		checkCols = append(checkCols, checkCol)
+	}
+	return checkCols, nil
+}
+
+// pushWeightStringForDistinct adds a weight_string projection
+func pushWeightStringForDistinct(ctx *plancontext.PlanningContext, plan logicalPlan, offset int) (newOffset int, err error) {
+	switch node := plan.(type) {
+	case *routeGen4:
+		allSelects := sqlparser.GetAllSelects(node.Select)
+		for _, sel := range allSelects {
+			expr, err := getWeightStringForSelectExpr(sel.SelectExprs[offset])
+			if err != nil {
+				return 0, err
+			}
+			if i := checkIfAlreadyExists(expr, sel, ctx.SemTable); i != -1 {
+				return i, nil
+			}
+			sel.SelectExprs = append(sel.SelectExprs, expr)
+			newOffset = len(sel.SelectExprs) - 1
+		}
+		// we leave the responsibility of truncating to distinct
+		node.eroute.TruncateColumnCount = 0
+	case *concatenateGen4:
+		for _, source := range node.sources {
+			newOffset, err = pushWeightStringForDistinct(ctx, source, offset)
+			if err != nil {
+				return 0, err
+			}
+		}
+		node.noNeedToTypeCheck = append(node.noNeedToTypeCheck, newOffset)
+	case *joinGen4:
+		lhsSolves := node.Left.ContainsTables()
+		rhsSolves := node.Right.ContainsTables()
+		expr := node.OutputColumns()[offset]
+		aliasedExpr, isAliased := expr.(*sqlparser.AliasedExpr)
+		if !isAliased {
+			return 0, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "cannot convert select expression to an aliased expression")
+		}
+		deps := ctx.SemTable.RecursiveDeps(aliasedExpr.Expr)
+		switch {
+		case deps.IsSolvedBy(lhsSolves):
+			offset, err = pushWeightStringForDistinct(ctx, node.Left, offset)
+			node.Cols = append(node.Cols, -(offset + 1))
+		case deps.IsSolvedBy(rhsSolves):
+			offset, err = pushWeightStringForDistinct(ctx, node.Right, offset)
+			node.Cols = append(node.Cols, offset+1)
+		default:
+			return 0, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "cannot push distinct weight string to both sides of the join")
+		}
+		newOffset = len(node.Cols) - 1
+	default:
+		return 0, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "bug: not supported pushWeightStringForDistinct on %T", plan)
+	}
+	return
+}
+
 func transformAndMerge(ctx *plancontext.PlanningContext, op *physical.Union) (sources []logicalPlan, err error) {
 	for i, source := range op.Sources {
 		// first we go over all the operator inputs and turn them into logical plans,
@@ -351,12 +439,12 @@ func transformAndMergeInOrder(ctx *plancontext.PlanningContext, op *physical.Uni
 }
 
 func createLogicalPlan(ctx *plancontext.PlanningContext, source abstract.PhysicalOperator, selStmt *sqlparser.Select) (logicalPlan, error) {
-	plan, err := transformToLogicalPlan(ctx, source)
+	plan, err := transformToLogicalPlan(ctx, source, false)
 	if err != nil {
 		return nil, err
 	}
 	if selStmt != nil {
-		plan, err = planHorizon(ctx, plan, selStmt)
+		plan, err = planHorizon(ctx, plan, selStmt, true)
 		if err != nil {
 			return nil, err
 		}
@@ -368,6 +456,7 @@ func createLogicalPlan(ctx *plancontext.PlanningContext, source abstract.Physica
 }
 
 func getCollationsFor(ctx *plancontext.PlanningContext, n *physical.Union) []collations.ID {
+	// TODO: coerce selects' select expressions' collations
 	var colls []collations.ID
 	for _, expr := range n.SelectStmts[0].SelectExprs {
 		aliasedE, ok := expr.(*sqlparser.AliasedExpr)
@@ -375,6 +464,11 @@ func getCollationsFor(ctx *plancontext.PlanningContext, n *physical.Union) []col
 			return nil
 		}
 		typ := ctx.SemTable.CollationForExpr(aliasedE.Expr)
+		if typ == collations.Unknown {
+			if t, hasT := ctx.SemTable.ExprTypes[aliasedE.Expr]; hasT && sqltypes.IsNumber(t.Type) {
+				typ = collations.CollationBinaryID
+			}
+		}
 		colls = append(colls, typ)
 	}
 	return colls
@@ -387,12 +481,12 @@ func transformDerivedPlan(ctx *plancontext.PlanningContext, op *physical.Derived
 	// expression containing our derived table's inner select and the derived
 	// table's alias.
 
-	plan, err := transformToLogicalPlan(ctx, op.Source)
+	plan, err := transformToLogicalPlan(ctx, op.Source, false)
 	if err != nil {
 		return nil, err
 	}
 
-	plan, err = planHorizon(ctx, plan, op.Query)
+	plan, err = planHorizon(ctx, plan, op.Query, false)
 	if err != nil {
 		return nil, err
 	}
diff --git a/go/vt/vtgate/planbuilder/ordered_aggregate.go b/go/vt/vtgate/planbuilder/ordered_aggregate.go
index 7506ccde07..4c091055dd 100644
--- a/go/vt/vtgate/planbuilder/ordered_aggregate.go
+++ b/go/vt/vtgate/planbuilder/ordered_aggregate.go
@@ -104,7 +104,7 @@ func (pb *primitiveBuilder) checkAggregates(sel *sqlparser.Select) error {
 		if hasAggregates {
 			return vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: cross-shard query with aggregates")
 		}
-		pb.plan = newDistinct(pb.plan, nil)
+		pb.plan = newDistinctV3(pb.plan)
 		return nil
 	}
 
diff --git a/go/vt/vtgate/planbuilder/subquery_op.go b/go/vt/vtgate/planbuilder/subquery_op.go
index a1abc30942..6191e5228c 100644
--- a/go/vt/vtgate/planbuilder/subquery_op.go
+++ b/go/vt/vtgate/planbuilder/subquery_op.go
@@ -25,18 +25,18 @@ import (
 )
 
 func transformSubQueryPlan(ctx *plancontext.PlanningContext, op *physical.SubQueryOp) (logicalPlan, error) {
-	innerPlan, err := transformToLogicalPlan(ctx, op.Inner)
+	innerPlan, err := transformToLogicalPlan(ctx, op.Inner, false)
 	if err != nil {
 		return nil, err
 	}
-	innerPlan, err = planHorizon(ctx, innerPlan, op.Extracted.Subquery.Select)
+	innerPlan, err = planHorizon(ctx, innerPlan, op.Extracted.Subquery.Select, true)
 	if err != nil {
 		return nil, err
 	}
 
 	argName := op.Extracted.GetArgName()
 	hasValuesArg := op.Extracted.GetHasValuesArg()
-	outerPlan, err := transformToLogicalPlan(ctx, op.Outer)
+	outerPlan, err := transformToLogicalPlan(ctx, op.Outer, false)
 
 	merged := mergeSubQueryOpPlan(ctx, innerPlan, outerPlan, op)
 	if merged != nil {
@@ -51,11 +51,11 @@ func transformSubQueryPlan(ctx *plancontext.PlanningContext, op *physical.SubQue
 }
 
 func transformCorrelatedSubQueryPlan(ctx *plancontext.PlanningContext, op *physical.CorrelatedSubQueryOp) (logicalPlan, error) {
-	outer, err := transformToLogicalPlan(ctx, op.Outer)
+	outer, err := transformToLogicalPlan(ctx, op.Outer, false)
 	if err != nil {
 		return nil, err
 	}
-	inner, err := transformToLogicalPlan(ctx, op.Inner)
+	inner, err := transformToLogicalPlan(ctx, op.Inner, false)
 	if err != nil {
 		return nil, err
 	}
diff --git a/go/vt/vtgate/planbuilder/union.go b/go/vt/vtgate/planbuilder/union.go
index c0305cdbc4..32dbf5ad21 100644
--- a/go/vt/vtgate/planbuilder/union.go
+++ b/go/vt/vtgate/planbuilder/union.go
@@ -78,7 +78,7 @@ func (pb *primitiveBuilder) processUnion(union *sqlparser.Union, reservedVars *s
 		}
 
 		if union.Distinct {
-			pb.plan = newDistinct(pb.plan, nil)
+			pb.plan = newDistinctV3(pb.plan)
 		}
 	}
 	pb.st.Outer = outer
diff --git a/go/vt/vttablet/onlineddl/vrepl.go b/go/vt/vttablet/onlineddl/vrepl.go
index 0fea74d6de..d2b1a445a1 100644
--- a/go/vt/vttablet/onlineddl/vrepl.go
+++ b/go/vt/vttablet/onlineddl/vrepl.go
@@ -81,11 +81,10 @@ func (v *VReplStream) isRunning() bool {
 
 // isFailed() returns true when the workflow has failed and will not retry
 func (v *VReplStream) isFailed() bool {
-	switch v.state {
-	case binlogplayer.BlpError:
+	switch {
+	case v.state == binlogplayer.BlpError:
 		return true
-	}
-	if strings.Contains(strings.ToLower(v.message), "error") {
+	case strings.Contains(strings.ToLower(v.message), "error"):
 		return true
 	}
 	return false
diff --git a/tools/build_version_flags.sh b/tools/build_version_flags.sh
index c9ea079d13..9b47322918 100755
--- a/tools/build_version_flags.sh
+++ b/tools/build_version_flags.sh
@@ -21,7 +21,7 @@ source $DIR/shell_functions.inc
 # a tar ball might be used, which will prevent the git metadata from being available.
 # Should this be the case then allow environment variables to be used to source
 # this information instead.
-DEFAULT_BUILD_GIT_REV=$(git rev-parse --short HEAD)
+DEFAULT_BUILD_GIT_REV=$(git rev-parse HEAD)
 DEFAULT_BUILD_GIT_BRANCH=$(git rev-parse --abbrev-ref HEAD)
 
 echo "\


Test patch:
diff --git a/go/cmd/vttlstest/vttlstest.go b/go/cmd/vttlstest/vttlstest.go
index dfa2f6b430..2606d99e42 100644
--- a/go/cmd/vttlstest/vttlstest.go
+++ b/go/cmd/vttlstest/vttlstest.go
@@ -36,8 +36,10 @@ vttlstest is a tool for generating test certificates and keys for TLS tests.
 To create a toplevel CA, use:
   $ vttlstest --root /tmp CreateCA
 
-To create an intermediate or leaf CA, use:
-  $ vttlstest --root /tmp CreateSignedCert servers
+To create an intermediate CA, use:
+  $ vttlstest --root /tmp CreateIntermediateCA servers
+
+To create a leaf certificate, use:
   $ vttlstest --root /tmp CreateSignedCert --parent servers server
 
 To get help on a command, use:
@@ -50,10 +52,11 @@ var cmdMap map[string]cmdFunc
 
 func init() {
 	cmdMap = map[string]cmdFunc{
-		"CreateCA":         cmdCreateCA,
-		"CreateCRL":        cmdCreateCRL,
-		"CreateSignedCert": cmdCreateSignedCert,
-		"RevokeCert":       cmdRevokeCert,
+		"CreateCA":             cmdCreateCA,
+		"CreateCRL":            cmdCreateCRL,
+		"CreateIntermediateCA": cmdCreateIntermediateCA,
+		"CreateSignedCert":     cmdCreateSignedCert,
+		"RevokeCert":           cmdRevokeCert,
 	}
 }
 
@@ -92,6 +95,24 @@ func cmdRevokeCert(subFlags *flag.FlagSet, args []string) {
 	tlstest.RevokeCertAndRegenerateCRL(*root, *parent, name)
 }
 
+func cmdCreateIntermediateCA(subFlags *flag.FlagSet, args []string) {
+	parent := subFlags.String("parent", "ca", "Parent cert name to use. Use 'ca' for the toplevel CA.")
+	serial := subFlags.String("serial", "01", "Serial number for the certificate to create. Should be different for two certificates with the same parent.")
+	commonName := subFlags.String("common_name", "", "Common name for the certificate. If empty, uses the name.")
+
+	subFlags.Parse(args)
+	if subFlags.NArg() != 1 {
+		log.Fatalf("CreateIntermediateCA command takes a single name as a parameter")
+	}
+
+	name := subFlags.Arg(0)
+	if *commonName == "" {
+		*commonName = name
+	}
+
+	tlstest.CreateIntermediateCA(*root, *parent, *serial, name, *commonName)
+}
+
 func cmdCreateSignedCert(subFlags *flag.FlagSet, args []string) {
 	parent := subFlags.String("parent", "ca", "Parent cert name to use. Use 'ca' for the toplevel CA.")
 	serial := subFlags.String("serial", "01", "Serial number for the certificate to create. Should be different for two certificates with the same parent.")
diff --git a/go/mysql/auth_server_clientcert_test.go b/go/mysql/auth_server_clientcert_test.go
index a6b2348b2b..b085fbb70a 100644
--- a/go/mysql/auth_server_clientcert_test.go
+++ b/go/mysql/auth_server_clientcert_test.go
@@ -107,7 +107,7 @@ func TestValidCert(t *testing.T) {
 		t.Errorf("userdata username is %v, expected %v", userData.Username, clientCertUsername)
 	}
 
-	expectedGroups := []string{"localhost", "127.0.0.1", clientCertUsername}
+	expectedGroups := []string{"localhost", clientCertUsername}
 	if !reflect.DeepEqual(userData.Groups, expectedGroups) {
 		t.Errorf("userdata groups is %v, expected %v", userData.Groups, expectedGroups)
 	}
diff --git a/go/test/endtoend/encryption/encryptedtransport/encrypted_transport_test.go b/go/test/endtoend/encryption/encryptedtransport/encrypted_transport_test.go
index b8d99532fb..fcd479fbad 100644
--- a/go/test/endtoend/encryption/encryptedtransport/encrypted_transport_test.go
+++ b/go/test/endtoend/encryption/encryptedtransport/encrypted_transport_test.go
@@ -54,6 +54,7 @@ to the go clients (for vtgate -> vttablet link). */
 package encryptedtransport
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	"io"
@@ -67,8 +68,6 @@ import (
 	"vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/vterrors"
 
-	"context"
-
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -259,16 +258,16 @@ func clusterSetUp(t *testing.T) (int, error) {
 	err := encryption.ExecuteVttlstestCommand("--root", certDirectory, "CreateCA")
 	require.NoError(t, err)
 
-	err = createSignedCert("ca", "01", "vttablet-server", "vttablet server CA")
+	err = createIntermediateCA("ca", "01", "vttablet-server", "vttablet server CA")
 	require.NoError(t, err)
 
-	err = createSignedCert("ca", "02", "vttablet-client", "vttablet client CA")
+	err = createIntermediateCA("ca", "02", "vttablet-client", "vttablet client CA")
 	require.NoError(t, err)
 
-	err = createSignedCert("ca", "03", "vtgate-server", "vtgate server CA")
+	err = createIntermediateCA("ca", "03", "vtgate-server", "vtgate server CA")
 	require.NoError(t, err)
 
-	err = createSignedCert("ca", "04", "vtgate-client", "vtgate client CA")
+	err = createIntermediateCA("ca", "04", "vtgate-client", "vtgate client CA")
 	require.NoError(t, err)
 
 	err = createSignedCert("vttablet-server", "01", "vttablet-server-instance", "vttablet server instance")
@@ -337,6 +336,19 @@ func clusterSetUp(t *testing.T) (int, error) {
 	return 0, nil
 }
 
+func createIntermediateCA(ca string, serial string, name string, commonName string) error {
+	log.Infof("Creating intermediate signed cert and key %s", commonName)
+	tmpProcess := exec.Command(
+		"vttlstest",
+		"--root", certDirectory,
+		"CreateIntermediateCA", "--",
+		"--parent", ca,
+		"--serial", serial,
+		"--common_name", commonName,
+		name)
+	return tmpProcess.Run()
+}
+
 func createSignedCert(ca string, serial string, name string, commonName string) error {
 	log.Infof("Creating signed cert and key %s", commonName)
 	tmpProcess := exec.Command(
diff --git a/go/test/endtoend/tabletmanager/custom_rule_topo_test.go b/go/test/endtoend/tabletmanager/custom_rule_topo_test.go
index f1112724d3..622f7003c5 100644
--- a/go/test/endtoend/tabletmanager/custom_rule_topo_test.go
+++ b/go/test/endtoend/tabletmanager/custom_rule_topo_test.go
@@ -78,15 +78,24 @@ func TestTopoCustomRule(t *testing.T) {
 	err = clusterInstance.VtctlclientProcess.ExecuteCommand("Validate")
 	require.Nil(t, err, "error should be Nil")
 
-	// Verify that query is working
-	result, err := vtctlExec("select id, value from t1", rTablet.Alias)
-	require.NoError(t, err)
-	resultMap := make(map[string]any)
-	err = json.Unmarshal([]byte(result), &resultMap)
-	require.NoError(t, err)
-
-	rowsAffected := resultMap["rows"].([]any)
-	assert.EqualValues(t, 2, len(rowsAffected))
+	// And wait until the query is working.
+	// We need a wait here because the instance we have created is a replica
+	// It might take a while to replicate the two rows.
+	timeout := time.Now().Add(10 * time.Second)
+	for time.Now().Before(timeout) {
+		result, err := vtctlExec("select id, value from t1", rTablet.Alias)
+		if err == nil {
+			resultMap := make(map[string]any)
+			err = json.Unmarshal([]byte(result), &resultMap)
+			require.NoError(t, err)
+
+			rowsAffected := resultMap["rows"].([]any)
+			if len(rowsAffected) == 2 {
+				break
+			}
+		}
+		time.Sleep(300 * time.Millisecond)
+	}
 
 	// Now update the topocustomrule file.
 	data = []byte(`[{
@@ -102,7 +111,7 @@ func TestTopoCustomRule(t *testing.T) {
 	require.Nil(t, err, "error should be Nil")
 
 	// And wait until the query fails with the right error.
-	timeout := time.Now().Add(10 * time.Second)
+	timeout = time.Now().Add(10 * time.Second)
 	for time.Now().Before(timeout) {
 		result, err := vtctlExec("select id, value from t1", rTablet.Alias)
 		if err != nil {
diff --git a/go/vt/servenv/buildinfo_test.go b/go/vt/servenv/buildinfo_test.go
index 31eb8b86c5..55617976fb 100644
--- a/go/vt/servenv/buildinfo_test.go
+++ b/go/vt/servenv/buildinfo_test.go
@@ -31,7 +31,7 @@ func TestVersionString(t *testing.T) {
 		buildUser:       "user",
 		buildTime:       now.Unix(),
 		buildTimePretty: "time is now",
-		buildGitRev:     "d54b87c",
+		buildGitRev:     "d54b87ca0be09b678bb4490060e8f23f890ddb92",
 		buildGitBranch:  "gitBranch",
 		goVersion:       "1.18",
 		goOS:            "amiga",
@@ -39,11 +39,11 @@ func TestVersionString(t *testing.T) {
 		version:         "v1.2.3-SNAPSHOT",
 	}
 
-	assert.Equal(t, "Version: v1.2.3-SNAPSHOT (Git revision d54b87c branch 'gitBranch') built on time is now by user@host using 1.18 amiga/amd64", v.String())
+	assert.Equal(t, "Version: v1.2.3-SNAPSHOT (Git revision d54b87ca0be09b678bb4490060e8f23f890ddb92 branch 'gitBranch') built on time is now by user@host using 1.18 amiga/amd64", v.String())
 
 	v.jenkinsBuildNumber = 422
 
-	assert.Equal(t, "Version: v1.2.3-SNAPSHOT (Jenkins build 422) (Git revision d54b87c branch 'gitBranch') built on time is now by user@host using 1.18 amiga/amd64", v.String())
+	assert.Equal(t, "Version: v1.2.3-SNAPSHOT (Jenkins build 422) (Git revision d54b87ca0be09b678bb4490060e8f23f890ddb92 branch 'gitBranch') built on time is now by user@host using 1.18 amiga/amd64", v.String())
 
 	assert.Equal(t, "5.7.9-vitess-v1.2.3-SNAPSHOT", v.MySQLVersion())
 }
diff --git a/go/vt/tlstest/tlstest.go b/go/vt/tlstest/tlstest.go
index 2ba14f600b..500a3974c4 100644
--- a/go/vt/tlstest/tlstest.go
+++ b/go/vt/tlstest/tlstest.go
@@ -19,108 +19,145 @@ limitations under the License.
 package tlstest
 
 import (
+	"bytes"
+	"crypto"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"errors"
 	"fmt"
+	"math/big"
+	"net"
 	"os"
-	"os/exec"
 	"path"
+	"strconv"
+	"time"
 
 	"vitess.io/vitess/go/vt/log"
 )
 
 const (
 	// CA is the name of the CA toplevel cert.
-	CA = "ca"
-
-	caConfigTemplate = `
-[ ca ]
-default_ca = default_ca
-
-[ default_ca ]
-database = %s
-default_md = sha256
-default_crl_days = 30
-
-[ req ]
- default_bits           = 4096
- default_keyfile        = keyfile.pem
- distinguished_name     = req_distinguished_name
- attributes             = req_attributes
- x509_extensions       = req_ext
- prompt                 = no
- output_password        = mypass
-[ req_distinguished_name ]
- C                      = US
- ST                     = California
- L                      = Mountain View
- O                      = Vitessio
- OU                     = Vitess
- CN                     = CA
- emailAddress           = test@email.address
-[ req_attributes ]
- challengePassword      = A challenge password
-[ req_ext ]
- basicConstraints       = CA:TRUE
-
-`
-
-	certConfig = `
-[ req ]
- default_bits           = 4096
- default_keyfile        = keyfile.pem
- distinguished_name     = req_distinguished_name
- attributes             = req_attributes
- req_extensions        = req_ext
- prompt                 = no
- output_password        = mypass
-[ req_distinguished_name ]
- C                      = US
- ST                     = California
- L                      = Mountain View
- O                      = Vitessio
- OU                     = Vitess
- CN                     = %s
- emailAddress           = test@email.address
-[ req_attributes ]
- challengePassword      = A challenge password
-[ req_ext ]
- basicConstraints       = CA:TRUE
- subjectAltName         = @alternate_names
- extendedKeyUsage       =serverAuth,clientAuth
-[ alternate_names ]
- DNS.1                  = localhost
- DNS.2                  = 127.0.0.1
- DNS.3                  = %s
-`
+	CA          = "ca"
+	permissions = 0700
 )
 
-// openssl runs the openssl binary with the provided command.
-func openssl(argv ...string) {
-	cmd := exec.Command("openssl", argv...)
-	output, err := cmd.CombinedOutput()
+func loadCert(certPath string) (*x509.Certificate, error) {
+	certData, err := os.ReadFile(certPath)
 	if err != nil {
-		if len(output) > 0 {
-			log.Errorf("openssl %v returned:\n%v", argv, string(output))
-		}
-		log.Fatalf("openssl %v failed: %v", argv, err)
+		return nil, err
+	}
+
+	block, _ := pem.Decode(certData)
+	if block == nil {
+		return nil, errors.New("failed to parse certificate PEM")
+	}
+	return x509.ParseCertificate(block.Bytes)
+}
+
+func saveCert(certificate *x509.Certificate, certPath string) error {
+	out := &bytes.Buffer{}
+	err := pem.Encode(out, &pem.Block{Type: "CERTIFICATE", Bytes: certificate.Raw})
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(certPath, out.Bytes(), permissions)
+}
+
+func generateKey() (crypto.PrivateKey, error) {
+	return ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+}
+
+func loadKey(keyPath string) (crypto.PrivateKey, error) {
+	keyData, err := os.ReadFile(keyPath)
+	if err != nil {
+		return nil, err
+	}
+
+	block, _ := pem.Decode(keyData)
+	if block == nil {
+		return nil, errors.New("failed to parse key PEM")
+	}
+
+	switch block.Type {
+	case "PRIVATE KEY":
+		return x509.ParsePKCS8PrivateKey(block.Bytes)
+	case "RSA PRIVATE KEY":
+		return x509.ParsePKCS1PrivateKey(block.Bytes)
+	case "EC PRIVATE KEY":
+		return x509.ParseECPrivateKey(block.Bytes)
+	default:
+		return nil, fmt.Errorf("unknown private key format: %+v", block.Type)
 	}
 }
 
-// createKeyDBAndCAConfig creates a key database and ca config file
-// for the passed in CA (possibly an intermediate CA)
-func createKeyDBAndCAConfig(root, parent string) {
-	databasePath := path.Join(root, parent+"-keys.db")
-	if _, err := os.Stat(databasePath); os.IsNotExist(err) {
-		if err := os.WriteFile(databasePath, []byte{}, os.ModePerm); err != nil {
-			log.Fatalf("cannot write file %v: %v", databasePath, err)
-		}
+func saveKey(key crypto.PrivateKey, keyPath string) error {
+	keyData, err := x509.MarshalPKCS8PrivateKey(key)
+	if err != nil {
+		return err
+	}
+	out := &bytes.Buffer{}
+	err = pem.Encode(out, &pem.Block{Type: "PRIVATE KEY", Bytes: keyData})
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(keyPath, out.Bytes(), permissions)
+}
+
+// pubkey is an interface to get a public key from a private key
+// The Go specification for a private key defines that this always
+// exists, although there's no interface for it since it would break
+// backwards compatibility. See https://pkg.go.dev/crypto#PrivateKey
+type pubKey interface {
+	Public() crypto.PublicKey
+}
+
+func publicKey(priv crypto.PrivateKey) crypto.PublicKey {
+	return priv.(pubKey).Public()
+}
+
+func signCert(parent *x509.Certificate, parentPriv crypto.PrivateKey, certPub crypto.PublicKey, commonName string, serial int64, ca bool) (*x509.Certificate, error) {
+	keyUsage := x509.KeyUsageDigitalSignature
+	var extKeyUsage []x509.ExtKeyUsage
+	var dnsNames []string
+	var ipAddresses []net.IP
+
+	if ca {
+		keyUsage = keyUsage | x509.KeyUsageCRLSign | x509.KeyUsageCertSign
+	} else {
+		extKeyUsage = []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth}
+		dnsNames = []string{"localhost", commonName}
+		ipAddresses = []net.IP{net.ParseIP("127.0.0.1"), net.ParseIP("::1")}
+	}
+
+	template := x509.Certificate{
+		SerialNumber: big.NewInt(serial),
+		Subject: pkix.Name{
+			CommonName: commonName,
+		},
+		NotBefore:             time.Now().Add(-30 * time.Second),
+		NotAfter:              time.Now().Add(24 * time.Hour),
+		KeyUsage:              keyUsage,
+		ExtKeyUsage:           extKeyUsage,
+		BasicConstraintsValid: true,
+		IsCA:                  ca,
+		DNSNames:              dnsNames,
+		IPAddresses:           ipAddresses,
 	}
 
-	config := path.Join(root, parent+"-ca.config")
-	if _, err := os.Stat(config); os.IsNotExist(err) {
-		if err := os.WriteFile(config, []byte(fmt.Sprintf(caConfigTemplate, databasePath)), os.ModePerm); err != nil {
-			log.Fatalf("cannot write file %v: %v", config, err)
-		}
+	// No parent defined means we create a self signed one.
+	if parent == nil {
+		parent = &template
 	}
+
+	certificate, err := x509.CreateCertificate(rand.Reader, &template, parent, certPub, parentPriv)
+	if err != nil {
+		return nil, err
+	}
+	return x509.ParseCertificate(certificate)
 }
 
 // CreateCA creates the toplevel 'ca' certificate and key, and places it
@@ -128,23 +165,68 @@ func createKeyDBAndCAConfig(root, parent string) {
 // directory.
 func CreateCA(root string) {
 	log.Infof("Creating test root CA in %v", root)
-	key := path.Join(root, "ca-key.pem")
-	cert := path.Join(root, "ca-cert.pem")
-	config := path.Join(root, "ca-ca.config")
-	createKeyDBAndCAConfig(root, "ca")
-
-	openssl("genrsa", "-out", key)
-
-	openssl("req",
-		"-new",
-		"-x509",
-		"-days", "3600",
-		"-nodes",
-		"-batch",
-		"-sha256",
-		"-config", config,
-		"-key", key,
-		"-out", cert)
+	keyPath := path.Join(root, "ca-key.pem")
+	certPath := path.Join(root, "ca-cert.pem")
+
+	priv, err := generateKey()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = saveKey(priv, keyPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	ca, err := signCert(nil, priv, publicKey(priv), CA, 1, true)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = saveCert(ca, certPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+}
+
+func CreateIntermediateCA(root, parent, serial, name, commonName string) {
+	caKeyPath := path.Join(root, parent+"-key.pem")
+	caCertPath := path.Join(root, parent+"-cert.pem")
+	keyPath := path.Join(root, name+"-key.pem")
+	certPath := path.Join(root, name+"-cert.pem")
+
+	caKey, err := loadKey(caKeyPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+	caCert, err := loadCert(caCertPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	priv, err := generateKey()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = saveKey(priv, keyPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	serialNr, err := strconv.ParseInt(serial, 10, 64)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	intermediate, err := signCert(caCert, caKey, publicKey(priv), commonName, serialNr, true)
+	if err != nil {
+		log.Fatal(err)
+	}
+	err = saveCert(intermediate, certPath)
+	if err != nil {
+		log.Fatal(err)
+	}
 }
 
 // CreateSignedCert creates a new certificate signed by the provided parent,
@@ -152,74 +234,147 @@ func CreateCA(root string) {
 // name is the file name to use. Common Name is the certificate common name.
 func CreateSignedCert(root, parent, serial, name, commonName string) {
 	log.Infof("Creating signed cert and key %v", commonName)
-	caKey := path.Join(root, parent+"-key.pem")
-	caCert := path.Join(root, parent+"-cert.pem")
-	key := path.Join(root, name+"-key.pem")
-	cert := path.Join(root, name+"-cert.pem")
-	req := path.Join(root, name+"-req.pem")
-
-	config := path.Join(root, name+".config")
-	if err := os.WriteFile(config, []byte(fmt.Sprintf(certConfig, commonName, commonName)), os.ModePerm); err != nil {
-		log.Fatalf("cannot write file %v: %v", config, err)
-	}
-	openssl("req",
-		"-newkey", "rsa:2048",
-		"-days", "3600",
-		"-nodes",
-		"-batch",
-		"-sha256",
-		"-config", config,
-		"-keyout", key,
-		"-out", req)
-	openssl("rsa", "-in", key, "-out", key)
-	openssl("x509", "-req",
-		"-in", req,
-		"-sha256",
-		"-days", "3600",
-		"-CA", caCert,
-		"-CAkey", caKey,
-		"-set_serial", serial,
-		"-extensions", "req_ext",
-		"-extfile", config,
-		"-out", cert)
+
+	caKeyPath := path.Join(root, parent+"-key.pem")
+	caCertPath := path.Join(root, parent+"-cert.pem")
+	keyPath := path.Join(root, name+"-key.pem")
+	certPath := path.Join(root, name+"-cert.pem")
+
+	caKey, err := loadKey(caKeyPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+	caCert, err := loadCert(caCertPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	priv, err := generateKey()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = saveKey(priv, keyPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	serialNr, err := strconv.ParseInt(serial, 10, 64)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	leaf, err := signCert(caCert, caKey, publicKey(priv), commonName, serialNr, false)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = saveCert(leaf, certPath)
+	if err != nil {
+		log.Fatal(err)
+	}
 }
 
 // CreateCRL creates a new empty certificate revocation list
 // for the provided parent
 func CreateCRL(root, parent string) {
 	log.Infof("Creating CRL for root CA in %v", root)
-	caKey := path.Join(root, parent+"-key.pem")
-	caCert := path.Join(root, parent+"-cert.pem")
-	configPath := path.Join(root, parent+"-ca.config")
+	caKeyPath := path.Join(root, parent+"-key.pem")
+	caCertPath := path.Join(root, parent+"-cert.pem")
 	crlPath := path.Join(root, parent+"-crl.pem")
-	createKeyDBAndCAConfig(root, parent)
-
-	openssl("ca", "-gencrl",
-		"-keyfile", caKey,
-		"-cert", caCert,
-		"-config", configPath,
-		"-out",
-		crlPath,
-	)
+
+	caKey, err := loadKey(caKeyPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+	caCert, err := loadCert(caCertPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	crlList, err := x509.CreateRevocationList(rand.Reader, &x509.RevocationList{
+		RevokedCertificates: nil,
+		Number:              big.NewInt(1),
+	}, caCert, caKey.(crypto.Signer))
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	out := &bytes.Buffer{}
+	err = pem.Encode(out, &pem.Block{Type: "X509 CRL", Bytes: crlList})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = os.WriteFile(crlPath, out.Bytes(), permissions)
+	if err != nil {
+		log.Fatal(err)
+	}
 }
 
 // RevokeCertAndRegenerateCRL revokes a provided certificate under the
 // provided parent CA and regenerates the CRL file for that parent
 func RevokeCertAndRegenerateCRL(root, parent, name string) {
 	log.Infof("Revoking certificate %s", name)
-	caKey := path.Join(root, parent+"-key.pem")
-	caCert := path.Join(root, parent+"-cert.pem")
-	cert := path.Join(root, name+"-cert.pem")
-	configPath := path.Join(root, parent+"-ca.config")
-	createKeyDBAndCAConfig(root, parent)
-
-	openssl("ca", "-revoke", cert,
-		"-keyfile", caKey,
-		"-cert", caCert,
-		"-config", configPath,
-	)
-
-	CreateCRL(root, parent)
+	caKeyPath := path.Join(root, parent+"-key.pem")
+	caCertPath := path.Join(root, parent+"-cert.pem")
+	crlPath := path.Join(root, parent+"-crl.pem")
+	certPath := path.Join(root, name+"-cert.pem")
+
+	certificate, err := loadCert(certPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// Check if CRL already exists. If it doesn't,
+	// create an empty CRL to start with.
+	_, err = os.Stat(crlPath)
+	if errors.Is(err, os.ErrNotExist) {
+		CreateCRL(root, parent)
+	}
+
+	data, err := os.ReadFile(crlPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+	crlList, err := x509.ParseCRL(data)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	revoked := crlList.TBSCertList.RevokedCertificates
+	revoked = append(revoked, pkix.RevokedCertificate{
+		SerialNumber:   certificate.SerialNumber,
+		RevocationTime: time.Now(),
+	})
+
+	caKey, err := loadKey(caKeyPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+	caCert, err := loadCert(caCertPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	newCrl, err := x509.CreateRevocationList(rand.Reader, &x509.RevocationList{
+		RevokedCertificates: revoked,
+		Number:              big.NewInt(int64(crlList.TBSCertList.Version) + 1),
+	}, caCert, caKey.(crypto.Signer))
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	out := &bytes.Buffer{}
+	err = pem.Encode(out, &pem.Block{Type: "X509 CRL", Bytes: newCrl})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = os.WriteFile(crlPath, out.Bytes(), permissions)
+	if err != nil {
+		log.Fatal(err)
+	}
 }
 
 // ClientServerKeyPairs is used in tests
@@ -272,12 +427,12 @@ func CreateClientServerCertPairs(root string) ClientServerKeyPairs {
 	revokedClientCertName := fmt.Sprintf("client-instance-%s", revokedClientSerial)
 	revokedClientCertCommonName := fmt.Sprintf("client%s.example.com", revokedClientSerial)
 
-	CreateSignedCert(root, CA, serverCASerial, serverCAName, serverCACommonName)
+	CreateIntermediateCA(root, CA, serverCASerial, serverCAName, serverCACommonName)
 	CreateSignedCert(root, serverCAName, serverSerial, serverCertName, serverCertCommonName)
 	CreateSignedCert(root, serverCAName, revokedServerSerial, revokedServerCertName, revokedServerCertCommonName)
 	RevokeCertAndRegenerateCRL(root, serverCAName, revokedServerCertName)
 
-	CreateSignedCert(root, CA, clientCASerial, clientCAName, clientCACommonName)
+	CreateIntermediateCA(root, CA, clientCASerial, clientCAName, clientCACommonName)
 	CreateSignedCert(root, clientCAName, clientCertSerial, clientCertName, clientCertCommonName)
 	CreateSignedCert(root, clientCAName, revokedClientSerial, revokedClientCertName, revokedClientCertCommonName)
 	RevokeCertAndRegenerateCRL(root, clientCAName, revokedClientCertName)
@@ -296,7 +451,7 @@ func CreateClientServerCertPairs(root string) ClientServerKeyPairs {
 		log.Fatalf("Could not read client CRL file")
 	}
 
-	err = os.WriteFile(combinedCRLPath, append(serverCRLBytes, clientCRLBytes...), 0777)
+	err = os.WriteFile(combinedCRLPath, append(serverCRLBytes, clientCRLBytes...), permissions)
 	if err != nil {
 		log.Fatalf("Could not write combined CRL file")
 	}
diff --git a/go/vt/vtadmin/api_test.go b/go/vt/vtadmin/api_test.go
index faa5795c43..887f8d4d8a 100644
--- a/go/vt/vtadmin/api_test.go
+++ b/go/vt/vtadmin/api_test.go
@@ -48,7 +48,6 @@ import (
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	tabletmanagerdatapb "vitess.io/vitess/go/vt/proto/tabletmanagerdata"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
-	"vitess.io/vitess/go/vt/proto/vschema"
 	vschemapb "vitess.io/vitess/go/vt/proto/vschema"
 	vtadminpb "vitess.io/vitess/go/vt/proto/vtadmin"
 	vtctldatapb "vitess.io/vitess/go/vt/proto/vtctldata"
@@ -2725,7 +2724,7 @@ func TestGetSrvVSchemas(t *testing.T) {
 							Id:   clusterID,
 							Name: clusterName,
 						},
-						SrvVSchema: &vschema.SrvVSchema{
+						SrvVSchema: &vschemapb.SrvVSchema{
 							Keyspaces: map[string]*vschemapb.Keyspace{
 								"commerce": {
 									Tables: map[string]*vschemapb.Table{
@@ -2762,7 +2761,7 @@ func TestGetSrvVSchemas(t *testing.T) {
 							Id:   clusterID,
 							Name: clusterName,
 						},
-						SrvVSchema: &vschema.SrvVSchema{
+						SrvVSchema: &vschemapb.SrvVSchema{
 							RoutingRules: &vschemapb.RoutingRules{},
 						},
 					},
@@ -2819,7 +2818,7 @@ func TestGetSrvVSchemas(t *testing.T) {
 							Id:   clusterID,
 							Name: clusterName,
 						},
-						SrvVSchema: &vschema.SrvVSchema{
+						SrvVSchema: &vschemapb.SrvVSchema{
 							RoutingRules: &vschemapb.RoutingRules{},
 						},
 					},
@@ -2860,7 +2859,7 @@ func TestGetSrvVSchemas(t *testing.T) {
 							Id:   clusterID,
 							Name: clusterName,
 						},
-						SrvVSchema: &vschema.SrvVSchema{
+						SrvVSchema: &vschemapb.SrvVSchema{
 							RoutingRules: &vschemapb.RoutingRules{},
 						},
 					},
@@ -2887,7 +2886,7 @@ func TestGetSrvVSchemas(t *testing.T) {
 							Id:   clusterID,
 							Name: clusterName,
 						},
-						SrvVSchema: &vschema.SrvVSchema{},
+						SrvVSchema: &vschemapb.SrvVSchema{},
 					},
 				},
 			},
diff --git a/go/vt/vtgate/engine/concatenate_test.go b/go/vt/vtgate/engine/concatenate_test.go
index 0acb217f27..bf1b54e061 100644
--- a/go/vt/vtgate/engine/concatenate_test.go
+++ b/go/vt/vtgate/engine/concatenate_test.go
@@ -39,8 +39,14 @@ func TestConcatenate_NoErrors(t *testing.T) {
 		inputs         []*sqltypes.Result
 		expectedResult *sqltypes.Result
 		expectedError  string
+		ignoreTypes    []int
 	}
 
+	r1 := r("id|col1|col2", "int64|varbinary|varbinary", "1|a1|b1", "2|a2|b2")
+	r2 := r("id|col1|col2", "int32|varbinary|varbinary", "1|a1|b1", "2|a2|b2")
+	combinedResult := r1
+	combinedResult.Rows = append(combinedResult.Rows, r2.Rows...)
+
 	testCases := []*testCase{{
 		testName: "empty results",
 		inputs: []*sqltypes.Result{
@@ -66,6 +72,14 @@ func TestConcatenate_NoErrors(t *testing.T) {
 			r("id|col3|col4", "int64|varchar|varbinary", "1|a1|b1", "2|a2|b2"),
 		},
 		expectedError: "merging field of different types is not supported",
+	}, {
+		testName: "ignored field types - ignored",
+		inputs: []*sqltypes.Result{
+			r("id|col1|col2", "int64|varbinary|varbinary", "1|a1|b1", "2|a2|b2"),
+			r("id|col1|col2", "int32|varbinary|varbinary", "1|a1|b1", "2|a2|b2"),
+		},
+		expectedResult: combinedResult,
+		ignoreTypes:    []int{0},
 	}, {
 		testName: "input source has different column count",
 		inputs: []*sqltypes.Result{
@@ -89,9 +103,8 @@ func TestConcatenate_NoErrors(t *testing.T) {
 			// input is added twice, since the first one is used by execute and the next by stream execute
 			sources = append(sources, &fakePrimitive{results: []*sqltypes.Result{input, input}})
 		}
-		concatenate := &Concatenate{
-			Sources: sources,
-		}
+
+		concatenate := NewConcatenate(sources, tc.ignoreTypes)
 
 		t.Run(tc.testName+"-Execute", func(t *testing.T) {
 			qr, err := concatenate.TryExecute(&noopVCursor{ctx: context.Background()}, nil, true)
@@ -121,13 +134,13 @@ func TestConcatenate_WithErrors(t *testing.T) {
 	strFailed := "failed"
 
 	fake := r("id|col1|col2", "int64|varchar|varbinary", "1|a1|b1", "2|a2|b2")
-	concatenate := &Concatenate{
-		Sources: []Primitive{
+	concatenate := NewConcatenate(
+		[]Primitive{
 			&fakePrimitive{results: []*sqltypes.Result{fake, fake}},
 			&fakePrimitive{results: []*sqltypes.Result{nil, nil}, sendErr: errors.New(strFailed)},
 			&fakePrimitive{results: []*sqltypes.Result{fake, fake}},
-		},
-	}
+		}, nil,
+	)
 	ctx := context.Background()
 	_, err := concatenate.TryExecute(&noopVCursor{ctx: ctx}, nil, true)
 	require.EqualError(t, err, strFailed)
@@ -135,13 +148,13 @@ func TestConcatenate_WithErrors(t *testing.T) {
 	_, err = wrapStreamExecute(concatenate, &noopVCursor{ctx: ctx}, nil, true)
 	require.EqualError(t, err, strFailed)
 
-	concatenate = &Concatenate{
-		Sources: []Primitive{
+	concatenate = NewConcatenate(
+		[]Primitive{
 			&fakePrimitive{results: []*sqltypes.Result{fake, fake}},
 			&fakePrimitive{results: []*sqltypes.Result{nil, nil}, sendErr: errors.New(strFailed)},
 			&fakePrimitive{results: []*sqltypes.Result{fake, fake}},
-		},
-	}
+		}, nil)
+
 	_, err = concatenate.TryExecute(&noopVCursor{ctx: ctx}, nil, true)
 	require.EqualError(t, err, strFailed)
 	_, err = wrapStreamExecute(concatenate, &noopVCursor{ctx: ctx}, nil, true)
diff --git a/go/vt/vtgate/engine/distinct_test.go b/go/vt/vtgate/engine/distinct_test.go
index 86652b7bf6..d5be097dd1 100644
--- a/go/vt/vtgate/engine/distinct_test.go
+++ b/go/vt/vtgate/engine/distinct_test.go
@@ -76,10 +76,27 @@ func TestDistinct(t *testing.T) {
 	}}
 
 	for _, tc := range testCases {
+		var checkCols []CheckCol
+		if len(tc.inputs.Rows) > 0 {
+			for i := range tc.inputs.Rows[0] {
+				collID := collations.Unknown
+				if tc.collations != nil {
+					collID = tc.collations[i]
+				}
+				if sqltypes.IsNumber(tc.inputs.Fields[i].Type) {
+					collID = collations.CollationBinaryID
+				}
+				checkCols = append(checkCols, CheckCol{
+					Col:       i,
+					Collation: collID,
+				})
+			}
+		}
 		t.Run(tc.testName+"-Execute", func(t *testing.T) {
 			distinct := &Distinct{
-				Source:        &fakePrimitive{results: []*sqltypes.Result{tc.inputs}},
-				ColCollations: tc.collations,
+				Source:    &fakePrimitive{results: []*sqltypes.Result{tc.inputs}},
+				CheckCols: checkCols,
+				Truncate:  false,
 			}
 
 			qr, err := distinct.TryExecute(&noopVCursor{ctx: context.Background()}, nil, true)
@@ -94,8 +111,8 @@ func TestDistinct(t *testing.T) {
 		})
 		t.Run(tc.testName+"-StreamExecute", func(t *testing.T) {
 			distinct := &Distinct{
-				Source:        &fakePrimitive{results: []*sqltypes.Result{tc.inputs}},
-				ColCollations: tc.collations,
+				Source:    &fakePrimitive{results: []*sqltypes.Result{tc.inputs}},
+				CheckCols: checkCols,
 			}
 
 			result, err := wrapStreamExecute(distinct, &noopVCursor{ctx: context.Background()}, nil, true)
@@ -111,3 +128,37 @@ func TestDistinct(t *testing.T) {
 		})
 	}
 }
+
+func TestWeightStringFallBack(t *testing.T) {
+	offsetOne := 1
+	checkCols := []CheckCol{{
+		Col:       0,
+		WsCol:     &offsetOne,
+		Collation: collations.Unknown,
+	}}
+	input := r("myid|weightstring(myid)",
+		"varchar|varbinary",
+		"monkey|monkey",
+		"horse|horse",
+		"horse|horse")
+
+	distinct := &Distinct{
+		Source:    &fakePrimitive{results: []*sqltypes.Result{input}},
+		CheckCols: checkCols,
+		Truncate:  true,
+	}
+
+	qr, err := distinct.TryExecute(&noopVCursor{ctx: context.Background()}, nil, true)
+	require.NoError(t, err)
+
+	got := fmt.Sprintf("%v", qr.Rows)
+	expected := fmt.Sprintf("%v", r("myid", "varchar", "monkey", "horse").Rows)
+	utils.MustMatch(t, expected, got)
+
+	// the primitive must not change just because one run needed weight strings
+	utils.MustMatch(t, []CheckCol{{
+		Col:       0,
+		WsCol:     &offsetOne,
+		Collation: collations.Unknown,
+	}}, distinct.CheckCols, "checkCols should not be updated")
+}
diff --git a/go/vt/vtgate/executor_select_test.go b/go/vt/vtgate/executor_select_test.go
index b379650bd6..e09212c4c4 100644
--- a/go/vt/vtgate/executor_select_test.go
+++ b/go/vt/vtgate/executor_select_test.go
@@ -881,12 +881,33 @@ func testRowCount(t *testing.T, executor *Executor, wantRowCount int64) {
 }
 
 func TestSelectLastInsertIdInUnion(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createExecutorEnv()
 	executor.normalize = true
-	sql := "select last_insert_id() as id union select id from user"
-	_, err := executorExec(executor, sql, map[string]*querypb.BindVariable{})
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "text type with an unknown/unsupported collation cannot be hashed")
+	primarySession.LastInsertId = 52
+
+	result1 := []*sqltypes.Result{{
+		Fields: []*querypb.Field{
+			{Name: "id", Type: sqltypes.Int32},
+		},
+		InsertID: 0,
+		Rows: [][]sqltypes.Value{{
+			sqltypes.NewInt32(52),
+		}},
+	}}
+	sbc1.SetResults(result1)
+
+	sql := "select last_insert_id() as id union select last_insert_id() as id"
+	got, err := executorExec(executor, sql, map[string]*querypb.BindVariable{})
+	require.NoError(t, err)
+	wantResult := &sqltypes.Result{
+		Fields: []*querypb.Field{
+			{Name: "id", Type: sqltypes.Int32},
+		},
+		Rows: [][]sqltypes.Value{{
+			sqltypes.NewInt32(52),
+		}},
+	}
+	utils.MustMatch(t, wantResult, got, "mismatch")
 }
 
 func TestSelectLastInsertIdInWhere(t *testing.T) {
diff --git a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
index 853b18d915..0e342f6d61 100644
--- a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
@@ -2146,7 +2146,6 @@ Gen4 plan same as above
         "FieldQuery": "select count(*) from (select `user`.col, user_extra.extra, weight_string(user_extra.extra) from `user`, user_extra where 1 != 1) as a where 1 != 1",
         "OrderBy": "(1|2) ASC",
         "Query": "select count(*) from (select `user`.col, user_extra.extra, weight_string(user_extra.extra) from `user`, user_extra where `user`.id = user_extra.user_id order by user_extra.extra asc) as a",
-        "ResultColumns": 2,
         "Table": "`user`, user_extra"
       }
     ]
@@ -2175,7 +2174,6 @@ Gen4 plan same as above
         "FieldQuery": "select col, count(*) from (select `user`.col, user_extra.extra, weight_string(user_extra.extra) from `user`, user_extra where 1 != 1) as a where 1 != 1 group by col",
         "OrderBy": "(1|2) ASC, 0 ASC",
         "Query": "select col, count(*) from (select `user`.col, user_extra.extra, weight_string(user_extra.extra) from `user`, user_extra where `user`.id = user_extra.user_id order by user_extra.extra asc) as a group by col order by col asc",
-        "ResultColumns": 2,
         "Table": "`user`, user_extra"
       }
     ]
diff --git a/go/vt/vtgate/planbuilder/testdata/select_cases.txt b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
index c1245a8c17..652ad343b9 100644
--- a/go/vt/vtgate/planbuilder/testdata/select_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
@@ -3021,3 +3021,92 @@ Gen4 plan same as above
   }
 }
 Gen4 plan same as above
+
+"select (select id from user order by id limit 1) from user_extra"
+{
+  "QueryType": "SELECT",
+  "Original": "select (select id from user order by id limit 1) from user_extra",
+  "Instructions": {
+    "OperatorType": "Subquery",
+    "Variant": "PulloutValue",
+    "PulloutVars": [
+      "__sq_has_values1",
+      "__sq1"
+    ],
+    "Inputs": [
+      {
+        "OperatorType": "Limit",
+        "Count": "INT64(1)",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "Scatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1",
+            "OrderBy": "(0|1) ASC",
+            "Query": "select id, weight_string(id) from `user` order by id asc limit :__upper_limit",
+            "ResultColumns": 1,
+            "Table": "`user`"
+          }
+        ]
+      },
+      {
+        "OperatorType": "Route",
+        "Variant": "Scatter",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select :__sq1 from user_extra where 1 != 1",
+        "Query": "select :__sq1 from user_extra",
+        "Table": "user_extra"
+      }
+    ]
+  }
+}
+{
+  "QueryType": "SELECT",
+  "Original": "select (select id from user order by id limit 1) from user_extra",
+  "Instructions": {
+    "OperatorType": "Subquery",
+    "Variant": "PulloutValue",
+    "PulloutVars": [
+      "__sq1"
+    ],
+    "Inputs": [
+      {
+        "OperatorType": "Limit",
+        "Count": "INT64(1)",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "Scatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1",
+            "OrderBy": "(0|1) ASC",
+            "Query": "select id, weight_string(id) from `user` order by id asc limit :__upper_limit",
+            "ResultColumns": 1,
+            "Table": "`user`"
+          }
+        ]
+      },
+      {
+        "OperatorType": "Route",
+        "Variant": "Scatter",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select :__sq1 from user_extra where 1 != 1",
+        "Query": "select :__sq1 from user_extra",
+        "Table": "user_extra"
+      }
+    ]
+  }
+}
diff --git a/go/vt/vtgate/planbuilder/testdata/union_cases.txt b/go/vt/vtgate/planbuilder/testdata/union_cases.txt
index df3b63e565..60c1531f1e 100644
--- a/go/vt/vtgate/planbuilder/testdata/union_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/union_cases.txt
@@ -60,6 +60,10 @@ Gen4 plan same as above
   "Original": "select id from user union select id from music",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:1)"
+    ],
+    "ResultColumns": 1,
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -68,8 +72,8 @@ Gen4 plan same as above
           "Name": "user",
           "Sharded": true
         },
-        "FieldQuery": "select id from `user` where 1 != 1 union select id from music where 1 != 1",
-        "Query": "select id from `user` union select id from music",
+        "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1 union select id, weight_string(id) from music where 1 != 1",
+        "Query": "select id, weight_string(id) from `user` union select id, weight_string(id) from music",
         "Table": "`user`"
       }
     ]
@@ -531,6 +535,10 @@ Gen4 plan same as above
   "Original": "(select id from user union select id from music) union select 1 from dual",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:1)"
+    ],
+    "ResultColumns": 1,
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -542,8 +550,8 @@ Gen4 plan same as above
               "Name": "user",
               "Sharded": true
             },
-            "FieldQuery": "select id from `user` where 1 != 1 union select id from music where 1 != 1",
-            "Query": "select id from `user` union select id from music",
+            "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1 union select id, weight_string(id) from music where 1 != 1",
+            "Query": "select id, weight_string(id) from `user` union select id, weight_string(id) from music",
             "Table": "`user`"
           },
           {
@@ -553,8 +561,8 @@ Gen4 plan same as above
               "Name": "main",
               "Sharded": false
             },
-            "FieldQuery": "select 1 from dual where 1 != 1",
-            "Query": "select distinct 1 from dual",
+            "FieldQuery": "select 1, weight_string(1) from dual where 1 != 1",
+            "Query": "select distinct 1, weight_string(1) from dual",
             "Table": "dual"
           }
         ]
@@ -822,6 +830,9 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "select 1 from music where id = 1 union select 1 from music where id = 2",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "0: binary"
+    ],
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -909,6 +920,10 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "(select id from user order by 1 desc) union (select id from user order by 1 asc)",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:1)"
+    ],
+    "ResultColumns": 1,
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -970,6 +985,9 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "select 1 union select null union select 1.0 union select '1' union select 2 union select 2.0 from user",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "0: binary"
+    ],
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -1064,6 +1082,11 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "(select user.id, user.name from user join user_extra where user_extra.extra = 'asdf') union select 'b','c' from user",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:2)",
+      "(1:3)"
+    ],
+    "ResultColumns": 2,
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -1071,7 +1094,7 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
           {
             "OperatorType": "Join",
             "Variant": "Join",
-            "JoinColumnIndexes": "L:0,L:1",
+            "JoinColumnIndexes": "L:0,L:1,L:2,L:3",
             "TableName": "`user`_user_extra",
             "Inputs": [
               {
@@ -1081,8 +1104,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
                   "Name": "user",
                   "Sharded": true
                 },
-                "FieldQuery": "select `user`.id, `user`.`name` from `user` where 1 != 1",
-                "Query": "select `user`.id, `user`.`name` from `user`",
+                "FieldQuery": "select `user`.id, `user`.`name`, weight_string(`user`.id), weight_string(`user`.`name`) from `user` where 1 != 1",
+                "Query": "select `user`.id, `user`.`name`, weight_string(`user`.id), weight_string(`user`.`name`) from `user`",
                 "Table": "`user`"
               },
               {
@@ -1105,8 +1128,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
               "Name": "user",
               "Sharded": true
             },
-            "FieldQuery": "select 'b', 'c' from `user` where 1 != 1",
-            "Query": "select distinct 'b', 'c' from `user`",
+            "FieldQuery": "select 'b', 'c', weight_string('b'), weight_string('c') from `user` where 1 != 1",
+            "Query": "select distinct 'b', 'c', weight_string('b'), weight_string('c') from `user`",
             "Table": "`user`"
           }
         ]
@@ -1177,6 +1200,11 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "select 'b','c' from user union (select user.id, user.name from user join user_extra where user_extra.extra = 'asdf')",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:2)",
+      "(1:3)"
+    ],
+    "ResultColumns": 2,
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -1188,14 +1216,14 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
               "Name": "user",
               "Sharded": true
             },
-            "FieldQuery": "select 'b', 'c' from `user` where 1 != 1",
-            "Query": "select distinct 'b', 'c' from `user`",
+            "FieldQuery": "select 'b', 'c', weight_string('b'), weight_string('c') from `user` where 1 != 1",
+            "Query": "select distinct 'b', 'c', weight_string('b'), weight_string('c') from `user`",
             "Table": "`user`"
           },
           {
             "OperatorType": "Join",
             "Variant": "Join",
-            "JoinColumnIndexes": "L:0,L:1",
+            "JoinColumnIndexes": "L:0,L:1,L:2,L:3",
             "TableName": "`user`_user_extra",
             "Inputs": [
               {
@@ -1205,8 +1233,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
                   "Name": "user",
                   "Sharded": true
                 },
-                "FieldQuery": "select `user`.id, `user`.`name` from `user` where 1 != 1",
-                "Query": "select `user`.id, `user`.`name` from `user`",
+                "FieldQuery": "select `user`.id, `user`.`name`, weight_string(`user`.id), weight_string(`user`.`name`) from `user` where 1 != 1",
+                "Query": "select `user`.id, `user`.`name`, weight_string(`user`.id), weight_string(`user`.`name`) from `user`",
                 "Table": "`user`"
               },
               {
@@ -1285,6 +1313,9 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "select count(*) as s from user union select count(*) as s from music",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "0: binary"
+    ],
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -1409,6 +1440,9 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
     "Inputs": [
       {
         "OperatorType": "Distinct",
+        "Collations": [
+          "(0:1)"
+        ],
         "Inputs": [
           {
             "OperatorType": "Route",
@@ -1417,8 +1451,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
               "Name": "user",
               "Sharded": true
             },
-            "FieldQuery": "select id from `user` where 1 != 1 union all select id + 1 from `user` where 1 != 1 union select user_id from user_extra where 1 != 1",
-            "Query": "select id from `user` union all select id + 1 from `user` union select user_id from user_extra",
+            "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1 union all select id + 1, weight_string(id + 1) from `user` where 1 != 1 union select user_id, weight_string(user_id) from user_extra where 1 != 1",
+            "Query": "select id, weight_string(id) from `user` union all select id + 1, weight_string(id + 1) from `user` union select user_id, weight_string(user_id) from user_extra",
             "Table": "`user`"
           }
         ]
@@ -1435,6 +1469,10 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "(select id from user union select id from music order by id) union select 1 from unsharded",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:1)"
+    ],
+    "ResultColumns": 1,
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -1446,8 +1484,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
               "Name": "user",
               "Sharded": true
             },
-            "FieldQuery": "select id from `user` where 1 != 1 union select id from music where 1 != 1",
-            "Query": "select id from `user` union select id from music",
+            "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1 union select id, weight_string(id) from music where 1 != 1",
+            "Query": "select id, weight_string(id) from `user` union select id, weight_string(id) from music",
             "Table": "`user`"
           },
           {
@@ -1457,8 +1495,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
               "Name": "main",
               "Sharded": false
             },
-            "FieldQuery": "select 1 from unsharded where 1 != 1",
-            "Query": "select distinct 1 from unsharded",
+            "FieldQuery": "select 1, weight_string(1) from unsharded where 1 != 1",
+            "Query": "select distinct 1, weight_string(1) from unsharded",
             "Table": "unsharded"
           }
         ]
@@ -1520,6 +1558,10 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
     "Inputs": [
       {
         "OperatorType": "Distinct",
+        "Collations": [
+          "(0:1)"
+        ],
+        "ResultColumns": 1,
         "Inputs": [
           {
             "OperatorType": "Concatenate",
@@ -1531,8 +1573,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
                   "Name": "user",
                   "Sharded": true
                 },
-                "FieldQuery": "select id from `user` where 1 != 1",
-                "Query": "select distinct id from `user` limit :__upper_limit",
+                "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1",
+                "Query": "select distinct id, weight_string(id) from `user` limit :__upper_limit",
                 "Table": "`user`"
               },
               {
@@ -1542,8 +1584,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
                   "Name": "main",
                   "Sharded": false
                 },
-                "FieldQuery": "select 3 from dual where 1 != 1",
-                "Query": "select distinct 3 from dual limit :__upper_limit",
+                "FieldQuery": "select 3, weight_string(3) from dual where 1 != 1",
+                "Query": "select distinct 3, weight_string(3) from dual limit :__upper_limit",
                 "Table": "dual"
               }
             ]
@@ -1658,6 +1700,10 @@ Gen4 plan same as above
   "Original": "select col from unsharded union select id from user union select col2 from unsharded union select col from user_extra",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:1)"
+    ],
+    "ResultColumns": 1,
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -1669,8 +1715,8 @@ Gen4 plan same as above
               "Name": "main",
               "Sharded": false
             },
-            "FieldQuery": "select col from unsharded where 1 != 1 union select col2 from unsharded where 1 != 1",
-            "Query": "select col from unsharded union select col2 from unsharded",
+            "FieldQuery": "select col, weight_string(col) from unsharded where 1 != 1 union select col2, weight_string(col2) from unsharded where 1 != 1",
+            "Query": "select col, weight_string(col) from unsharded union select col2, weight_string(col2) from unsharded",
             "Table": "unsharded"
           },
           {
@@ -1680,8 +1726,8 @@ Gen4 plan same as above
               "Name": "user",
               "Sharded": true
             },
-            "FieldQuery": "select id from `user` where 1 != 1 union select col from user_extra where 1 != 1",
-            "Query": "select id from `user` union select col from user_extra",
+            "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1 union select col, weight_string(col) from user_extra where 1 != 1",
+            "Query": "select id, weight_string(id) from `user` union select col, weight_string(col) from user_extra",
             "Table": "`user`"
           }
         ]
@@ -1808,11 +1854,14 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "SimpleProjection",
     "Columns": [
-      1
+      2
     ],
     "Inputs": [
       {
         "OperatorType": "Distinct",
+        "Collations": [
+          "(0:1)"
+        ],
         "Inputs": [
           {
             "OperatorType": "Concatenate",
@@ -1824,8 +1873,8 @@ Gen4 plan same as above
                   "Name": "user",
                   "Sharded": true
                 },
-                "FieldQuery": "select id + 42 as foo, 1 from `user` where 1 != 1",
-                "Query": "select distinct id + 42 as foo, 1 from `user`",
+                "FieldQuery": "select id + 42 as foo, weight_string(id + 42), 1 from `user` where 1 != 1",
+                "Query": "select distinct id + 42 as foo, weight_string(id + 42), 1 from `user`",
                 "Table": "`user`"
               },
               {
@@ -1835,8 +1884,8 @@ Gen4 plan same as above
                   "Name": "main",
                   "Sharded": false
                 },
-                "FieldQuery": "select 1 + id as foo, 1 from unsharded where 1 != 1",
-                "Query": "select distinct 1 + id as foo, 1 from unsharded",
+                "FieldQuery": "select 1 + id as foo, weight_string(1 + id), 1 from unsharded where 1 != 1",
+                "Query": "select distinct 1 + id as foo, weight_string(1 + id), 1 from unsharded",
                 "Table": "unsharded"
               }
             ]