Repository: vitessio/vitess
Pull request #6001 (2020-04-01 11:32:33)

Vreplication flaky test: make vcopier test non-flaky
The copy-catchup-fast-forward phases in the vreplication flow have different timeouts that determine how long a phase may last. This can result in extra transactions that update pos. The flakiness in vcopier_test.go was because we were incorrectly expecting a specific sequence of sql statements in a couple of tests. 

Changing it to expect only the key sql statements and ignoring non-transaction queries has fixed the flakiness.

Additionally SecondsBehindMaster is now set correctly *after* the events are applied.


Patch:
diff --git a/.github/workflows/sonar_analysis.yml b/.github/workflows/sonar_analysis.yml
index 3381d93b22..f901499a85 100644
--- a/.github/workflows/sonar_analysis.yml
+++ b/.github/workflows/sonar_analysis.yml
@@ -2,7 +2,7 @@ name: sonar_analysis
 on:
   push:
     branches:
-      - 'sonartest'
+      - 'master'
 jobs:
 
   build:
diff --git a/Makefile b/Makefile
index 887dc72b3a..b47e3e7bd1 100644
--- a/Makefile
+++ b/Makefile
@@ -74,6 +74,14 @@ install: build
 	# binaries
 	mkdir -p "$${PREFIX}/bin"
 	cp "$${VTROOT}/bin/"{mysqlctld,vtctld,vtctlclient,vtgate,vttablet,vtworker,vtbackup} "$${PREFIX}/bin/"
+	# config files
+	mkdir -p "$${PREFIX}/src/vitess.io/vitess"
+	cp -R config "$${PREFIX}/src/vitess.io/vitess/"
+	# also symlink config files in the old location
+	ln -sf src/vitess.io/vitess/config "$${PREFIX}/config"
+	# vtctld web UI files
+	mkdir -p "$${PREFIX}/src/vitess.io/vitess/web/vtctld2"
+	cp -R web/vtctld2/app "$${PREFIX}/src/vitess.io/vitess/web/vtctld2/"
 
 parser:
 	make -C go/vt/sqlparser
@@ -281,6 +289,12 @@ docker_test:
 docker_unit_test:
 	go run test.go -flavor $(flavor) unit
 
+# This can be used to rebalance the total average runtime of each group of
+# tests in Travis. The results are saved in test/config.json, which you can
+# then commit and push.
+rebalance_tests:
+	go run test.go -rebalance 5
+
 # Release a version.
 # This will generate a tar.gz file into the releases folder with the current source
 release: docker_base
@@ -310,7 +324,7 @@ tools:
 
 minimaltools:
 	echo $$(date): Installing minimal dependencies
-	BUILD_CHROME=0 BUILD_JAVA=0 BUILD_CONSUL=0 ./bootstrap.sh
+	BUILD_JAVA=0 BUILD_CONSUL=0 ./bootstrap.sh
 
 dependency_check:
 	./tools/dependency_check.sh
diff --git a/bootstrap.sh b/bootstrap.sh
index a2f7891dd0..40e26ad7ce 100755
--- a/bootstrap.sh
+++ b/bootstrap.sh
@@ -25,7 +25,6 @@ source ./dev.env
 
 BUILD_JAVA=${BUILD_JAVA:-1}
 BUILD_CONSUL=${BUILD_CONSUL:-1}
-BUILD_CHROME=${BUILD_CHROME:-1}
 
 #
 # 0. Initialization and helper methods.
@@ -173,13 +172,13 @@ function install_k3s() {
 
   case $(uname) in
     Linux)  local platform=linux;;
-    *)   echo "WARNING: unsupported platform. K3s only supports running on Linux, the k8s topology will not be available for local examples."; return;;
+    *)   echo "ERROR: unsupported platform. K3s only supports running on Linux"; exit 1;;
   esac
 
   case $(get_arch) in
       aarch64)  local target="-arm64";;
       x86_64)  local target="";;
-      *)   echo "WARNING: unsupported architecture, the k8s topology will not be available for local examples."; return;;
+      *)   echo "ERROR: unsupported architecture"; exit 1;;
   esac
 
   download_url=https://github.com/rancher/k3s/releases/download
@@ -192,6 +191,7 @@ function install_k3s() {
 }
 command -v  k3s || install_dep "k3s" "v1.0.0" "$VTROOT/dist/k3s" install_k3s
 
+
 # Download and install consul, link consul binary into our root.
 function install_consul() {
   local version="$1"
@@ -218,40 +218,5 @@ if [ "$BUILD_CONSUL" == 1 ] ; then
   install_dep "Consul" "1.4.0" "$VTROOT/dist/consul" install_consul
 fi
 
-# Download chromedriver
-function install_chromedriver() {
-  local version="$1"
-  local dist="$2"
-
-  case $(uname) in
-    Linux)  local platform=linux;;
-    *)   echo "Platform not supported for vtctl-web tests. Skipping chromedriver install."; return;;
-  esac
-
-  if [ "$(arch)" == "aarch64" ] ; then
-      os=$(cat /etc/*release | grep "^ID=" | cut -d '=' -f 2)
-      case $os in
-          ubuntu|debian)
-              sudo apt-get update -y && sudo apt install -y --no-install-recommends unzip libglib2.0-0 libnss3 libx11-6
-	      ;;
-	  centos|fedora)
-	      sudo yum update -y && yum install -y libX11 unzip wget
-	      ;;
-      esac
-      echo "For Arm64, using prebuilt binary from electron (https://github.com/electron/electron/) of version 76.0.3809.126"
-      wget https://github.com/electron/electron/releases/download/v6.0.3/chromedriver-v6.0.3-linux-arm64.zip
-      unzip -o -q chromedriver-v6.0.3-linux-arm64.zip -d "$dist"
-      rm chromedriver-v6.0.3-linux-arm64.zip
-  else
-      curl -sL "https://chromedriver.storage.googleapis.com/$version/chromedriver_linux64.zip" > chromedriver_linux64.zip
-      unzip -o -q chromedriver_linux64.zip -d "$dist"
-      rm chromedriver_linux64.zip
-  fi
-}
-
-if [ "$BUILD_CHROME" == 1 ] ; then
-	install_dep "chromedriver" "73.0.3683.20" "$VTROOT/dist/chromedriver" install_chromedriver
-fi
-
 echo
 echo "bootstrap finished - run 'make build' to compile"
diff --git a/build.env b/build.env
index a09807bdf5..0cc095fe65 100755
--- a/build.env
+++ b/build.env
@@ -17,7 +17,7 @@
 source ./tools/shell_functions.inc
 
 go version >/dev/null 2>&1 || fail "Go is not installed or is not in \$PATH. See https://vitess.io/contributing/build-from-source for install instructions."
-goversion_min 1.13 || fail "Go version reported: `go version`. Version 1.13+ required. See https://vitess.io/contributing/build-from-source for install instructions."
+goversion_min 1.13 || fail "Go version reported: ${go version}. Version 1.13+ required. See https://vitess.io/contributing/build-from-source for install instructions."
 
 mkdir -p dist
 mkdir -p bin
diff --git a/docker/bootstrap/Dockerfile.common b/docker/bootstrap/Dockerfile.common
index 4ee1e190b9..6a3f8d7b20 100644
--- a/docker/bootstrap/Dockerfile.common
+++ b/docker/bootstrap/Dockerfile.common
@@ -14,6 +14,12 @@ RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-ins
     libtool \
     make \
     openjdk-8-jdk \
+    python-crypto \
+    python-dev \
+    python-mysqldb \
+    python-pip \
+    ruby \
+    ruby-dev \
     software-properties-common \
     virtualenv \
     unzip \
@@ -40,7 +46,9 @@ ENV USER vitess
 # Copy files needed for bootstrap
 COPY bootstrap.sh dev.env build.env go.mod go.sum /vt/src/vitess.io/vitess/
 COPY config /vt/src/vitess.io/vitess/config
+COPY third_party /vt/src/vitess.io/vitess/third_party
 COPY tools /vt/src/vitess.io/vitess/tools
+COPY travis /vt/src/vitess.io/vitess/travis
 
 # Create vitess user
 RUN groupadd -r vitess && useradd -r -g vitess vitess && \
diff --git a/examples/compose/vtcompose/vtcompose.go b/examples/compose/vtcompose/vtcompose.go
index eb9cd7f091..912081b66e 100644
--- a/examples/compose/vtcompose/vtcompose.go
+++ b/examples/compose/vtcompose/vtcompose.go
@@ -34,46 +34,23 @@ import (
 	"vitess.io/vitess/go/vt/log"
 )
 
-const (
-	DefaultWebPort        = 8080
-	webPortUsage          = "Web port to be used"
-	DefaultGrpcPort       = 15999
-	gRpcPortUsage         = "gRPC port to be used"
-	DefaultMysqlPort      = 15306
-	mySqlPortUsage        = "mySql port to be used"
-	DefaultKeyspaceData   = "test_keyspace:2:1:create_messages.sql,create_tokens.sql;unsharded_keyspace:0:0:create_dinosaurs.sql,create_eggs.sql"
-	keyspaceDataUsage     = "List of keyspace_name/external_db_name:num_of_shards:num_of_replica_tablets:schema_files:<optional>lookup_keyspace_name separated by ';'"
-	DefaultCell           = "test"
-	cellUsage             = "Vitess Cell name"
-	DefaultExternalDbData = ""
-	externalDbDataUsage   = "List of Data corresponding to external DBs. List of <external_db_name>,<DB_HOST>,<DB_PORT>,<DB_USER>,<DB_PASS>,<DB_CHARSET> separated by ';'"
-	DefaultTopologyFlags  = "-topo_implementation consul -topo_global_server_address consul1:8500 -topo_global_root vitess/global"
-	topologyFlagsUsage    = "Vitess Topology Flags config"
-)
-
 var (
 	tabletsUsed           = 0
 	tablesPath            = "tables/"
 	baseDockerComposeFile = flag.String("base_yaml", "vtcompose/docker-compose.base.yml", "Starting docker-compose yaml")
 	baseVschemaFile       = flag.String("base_vschema", "vtcompose/base_vschema.json", "Starting vschema json")
 
-	topologyFlags  = flag.String("topologyFlags", DefaultTopologyFlags, topologyFlagsUsage)
-	webPort        = flag.Int("webPort", DefaultWebPort, webPortUsage)
-	gRpcPort       = flag.Int("gRpcPort", DefaultGrpcPort, gRpcPortUsage)
-	mySqlPort      = flag.Int("mySqlPort", DefaultMysqlPort, mySqlPortUsage)
-	cell           = flag.String("cell", DefaultCell, cellUsage)
-	keyspaceData   = flag.String("keyspaceData", DefaultKeyspaceData, keyspaceDataUsage)
-	externalDbData = flag.String("externalDbData", DefaultExternalDbData, externalDbDataUsage)
+	topologyFlags = flag.String("topologyFlags",
+		"-topo_implementation consul -topo_global_server_address consul1:8500 -topo_global_root vitess/global",
+		"Vitess Topology Flags config")
+	webPort        = flag.String("webPort", "8080", "Web port to be used")
+	gRpcPort       = flag.String("gRpcPort", "15999", "gRPC port to be used")
+	mySqlPort      = flag.String("mySqlPort", "15306", "mySql port to be used")
+	cell           = flag.String("cell", "test", "Vitess Cell name")
+	keyspaceData   = flag.String("keyspaceData", "test_keyspace:2:1:create_messages.sql,create_tokens.sql;unsharded_keyspace:0:0:create_dinosaurs.sql,create_eggs.sql", "List of keyspace_name/external_db_name:num_of_shards:num_of_replica_tablets:schema_files:<optional>lookup_keyspace_name separated by ';'")
+	externalDbData = flag.String("externalDbData", "", "List of Data corresponding to external DBs. List of <external_db_name>,<DB_HOST>,<DB_PORT>,<DB_USER>,<DB_PASS>,<DB_CHARSET> separated by ';'")
 )
 
-type vtOptions struct {
-	webPort       int
-	gRpcPort      int
-	mySqlPort     int
-	topologyFlags string
-	cell          string
-}
-
 type keyspaceInfo struct {
 	keyspace        string
 	shards          int
@@ -93,20 +70,8 @@ type externalDbInfo struct {
 	dbCharset string
 }
 
-func newKeyspaceInfo(
-	keyspace string,
-	shards int,
-	replicaTablets int,
-	schemaFiles []string,
-	lookupKeyspace string,
-) keyspaceInfo {
-	k := keyspaceInfo{
-		keyspace:        keyspace,
-		shards:          shards,
-		replicaTablets:  replicaTablets,
-		schemaFileNames: schemaFiles,
-		lookupKeyspace:  lookupKeyspace,
-	}
+func newKeyspaceInfo(keyspace string, shards int, replicaTablets int, schemaFiles []string, lookupKeyspace string) keyspaceInfo {
+	k := keyspaceInfo{keyspace: keyspace, shards: shards, replicaTablets: replicaTablets, schemaFileNames: schemaFiles, lookupKeyspace: lookupKeyspace}
 	if len(strings.TrimSpace(lookupKeyspace)) == 0 {
 		k.useLookups = false
 	} else {
@@ -118,24 +83,21 @@ func newKeyspaceInfo(
 }
 
 func newExternalDbInfo(dbName, dbHost, dbPort, dbUser, dbPass, dbCharset string) externalDbInfo {
-	return externalDbInfo{
-		dbName:    dbName,
-		dbHost:    dbHost,
-		dbPort:    dbPort,
-		dbUser:    dbUser,
-		dbPass:    dbPass,
-		dbCharset: dbCharset,
-	}
+	d := externalDbInfo{dbName: dbName, dbHost: dbHost, dbPort: dbPort, dbUser: dbUser, dbPass: dbPass, dbCharset: dbCharset}
+	return d
 }
 
-func parseKeyspaceInfo(keyspaceData string) map[string]keyspaceInfo {
+func main() {
+	flag.Parse()
 	keyspaceInfoMap := make(map[string]keyspaceInfo)
+	externalDbInfoMap := make(map[string]externalDbInfo)
 
-	for _, v := range strings.Split(keyspaceData, ";") {
+	for _, v := range strings.Split(*keyspaceData, ";") {
 		tokens := strings.Split(v, ":")
 		shards, _ := strconv.Atoi(tokens[1])
 		replicaTablets, _ := strconv.Atoi(tokens[2])
 		schemaFileNames := strings.Split(tokens[3], ",")
+		print(shards)
 
 		if len(tokens) > 4 {
 			keyspaceInfoMap[tokens[0]] = newKeyspaceInfo(tokens[0], shards, replicaTablets, schemaFileNames, tokens[4])
@@ -144,39 +106,17 @@ func parseKeyspaceInfo(keyspaceData string) map[string]keyspaceInfo {
 		}
 	}
 
-	return keyspaceInfoMap
-}
-
-func parseExternalDbData(externalDbData string) map[string]externalDbInfo {
-	externalDbInfoMap := make(map[string]externalDbInfo)
-	for _, v := range strings.Split(externalDbData, ";") {
+	for _, v := range strings.Split(*externalDbData, ";") {
 		tokens := strings.Split(v, ":")
 		if len(tokens) > 1 {
-			externalDbInfoMap[tokens[0]] =
-				newExternalDbInfo(tokens[0], tokens[1], tokens[2], tokens[3], tokens[4], tokens[5])
+			externalDbInfoMap[tokens[0]] = newExternalDbInfo(tokens[0], tokens[1], tokens[2], tokens[3], tokens[4], tokens[5])
 		}
 	}
 
-	return externalDbInfoMap
-}
-
-func main() {
-	flag.Parse()
-	keyspaceInfoMap := parseKeyspaceInfo(*keyspaceData)
-	externalDbInfoMap := parseExternalDbData(*externalDbData)
-	vtOpts := vtOptions{
-		webPort:       *webPort,
-		gRpcPort:      *gRpcPort,
-		mySqlPort:     *mySqlPort,
-		topologyFlags: *topologyFlags,
-		cell:          *cell,
-	}
-
-	// Write schemaFile.
 	for k, v := range keyspaceInfoMap {
 		if _, ok := externalDbInfoMap[k]; !ok {
 			v.schemaFile = createFile(fmt.Sprintf("%s%s_schema_file.sql", tablesPath, v.keyspace))
-			appendToSqlFile(v.schemaFileNames, v.schemaFile)
+			appendtoSqlFile(v.schemaFileNames, v.schemaFile)
 			closeFile(v.schemaFile)
 		}
 	}
@@ -198,8 +138,8 @@ func main() {
 			vSchemaFile, primaryTableColumns = addTablesVschemaPatch(vSchemaFile, keyspaceData.schemaFileNames)
 
 			if keyspaceData.useLookups {
-				lookup := keyspaceInfoMap[keyspaceData.lookupKeyspace]
-				vSchemaFile = addLookupDataToVschema(vSchemaFile, lookup.schemaFileNames, primaryTableColumns, lookup.keyspace)
+				lookupKeyspace := keyspaceInfoMap[keyspaceData.lookupKeyspace]
+				vSchemaFile = addLookupDataToVschema(vSchemaFile, lookupKeyspace.schemaFileNames, primaryTableColumns, lookupKeyspace.keyspace)
 			}
 		}
 
@@ -208,7 +148,7 @@ func main() {
 
 	// Docker Compose File Patches
 	dockerComposeFile := readFile(*baseDockerComposeFile)
-	dockerComposeFile = applyDockerComposePatches(dockerComposeFile, keyspaceInfoMap, externalDbInfoMap, vtOpts)
+	dockerComposeFile = applyDockerComposePatches(dockerComposeFile, keyspaceInfoMap, externalDbInfoMap)
 	writeFile(dockerComposeFile, "docker-compose.yml")
 }
 
@@ -287,7 +227,7 @@ func handleError(err error) {
 	}
 }
 
-func appendToSqlFile(schemaFileNames []string, f *os.File) {
+func appendtoSqlFile(schemaFileNames []string, f *os.File) {
 	for _, file := range schemaFileNames {
 		data, err := ioutil.ReadFile(tablesPath + file)
 		_, err = f.Write(data)
@@ -333,6 +273,7 @@ func getKeyColumns(sqlFile string) string {
 
 	r, _ := regexp.Compile("[^PRIMARY] (KEY|UNIQUE KEY) .*\\((.*)\\).*")
 	rs := r.FindStringSubmatch(string(sqlFileData))
+	print(rs[2])
 	// replace all ` from column names if exists
 	return strings.ReplaceAll(rs[2], "`", "")
 }
@@ -351,12 +292,7 @@ func addTablesVschemaPatch(vSchemaFile []byte, schemaFileNames []string) ([]byte
 	return vSchemaFile, primaryTableColumns
 }
 
-func addLookupDataToVschema(
-	vSchemaFile []byte,
-	schemaFileNames []string,
-	primaryTableColumns map[string]string,
-	keyspace string,
-) []byte {
+func addLookupDataToVschema(vSchemaFile []byte, schemaFileNames []string, primaryTableColumns map[string]string, keyspace string) []byte {
 	for _, fileName := range schemaFileNames {
 		tableName := fileName[7 : len(fileName)-4]
 		lookupTableOwner := ""
@@ -375,8 +311,8 @@ func addLookupDataToVschema(
 		vSchemaFile = applyJsonInMemoryPatch(vSchemaFile, addToColumnVIndexes(lookupTableOwner, firstColumnName, tableName))
 
 		// Generate Vschema lookup hash types
-		lookupHash := generateVschemaLookupHash(tableName, keyspace, firstColumnName, primaryTableColumns[lookupTableOwner], lookupTableOwner)
-		vSchemaFile = applyJsonInMemoryPatch(vSchemaFile, lookupHash)
+		vSchemaFile = applyJsonInMemoryPatch(vSchemaFile,
+			generateVschemaLookupHash(tableName, keyspace, firstColumnName, primaryTableColumns[lookupTableOwner], lookupTableOwner))
 	}
 
 	return vSchemaFile
@@ -399,12 +335,7 @@ func writeFile(file []byte, fileName string) {
 	}
 }
 
-func applyKeyspaceDependentPatches(
-	dockerComposeFile []byte,
-	keyspaceData keyspaceInfo,
-	externalDbInfoMap map[string]externalDbInfo,
-	opts vtOptions,
-) []byte {
+func applyKeyspaceDependentPatches(dockerComposeFile []byte, keyspaceData keyspaceInfo, externalDbInfoMap map[string]externalDbInfo) []byte {
 	var externalDbInfo externalDbInfo
 	if val, ok := externalDbInfoMap[keyspaceData.keyspace]; ok {
 		externalDbInfo = val
@@ -423,13 +354,12 @@ func applyKeyspaceDependentPatches(
 		masterTablets = append(masterTablets, strconv.Itoa((i+1)*100+1))
 	}
 
-	schemaLoad := generateSchemaload(masterTablets, "", keyspaceData.keyspace, externalDbInfo, opts)
-	dockerComposeFile = applyInMemoryPatch(dockerComposeFile, schemaLoad)
+	dockerComposeFile = applyInMemoryPatch(dockerComposeFile, generateSchemaload(masterTablets, "", keyspaceData.keyspace, externalDbInfo))
 
 	// Append Master and Replica Tablets
 	if keyspaceData.shards < 2 {
 		tabAlias = tabAlias + 100
-		dockerComposeFile = applyTabletPatches(dockerComposeFile, tabAlias, shard, keyspaceData, externalDbInfoMap, opts)
+		dockerComposeFile = applyTabletPatches(dockerComposeFile, tabAlias, shard, keyspaceData, externalDbInfoMap)
 	} else {
 		// Determine shard range
 		for i := 0; i < keyspaceData.shards; i++ {
@@ -441,7 +371,7 @@ func applyKeyspaceDependentPatches(
 				shard = fmt.Sprintf("%x-%x", interval*(i), interval*(i+1))
 			}
 			tabAlias = tabAlias + 100
-			dockerComposeFile = applyTabletPatches(dockerComposeFile, tabAlias, shard, keyspaceData, externalDbInfoMap, opts)
+			dockerComposeFile = applyTabletPatches(dockerComposeFile, tabAlias, shard, keyspaceData, externalDbInfoMap)
 		}
 	}
 
@@ -449,99 +379,90 @@ func applyKeyspaceDependentPatches(
 	return dockerComposeFile
 }
 
-func applyDefaultDockerPatches(dockerComposeFile []byte, opts vtOptions) []byte {
-	dockerComposeFile = applyInMemoryPatch(dockerComposeFile, generateVtctld(opts))
-	dockerComposeFile = applyInMemoryPatch(dockerComposeFile, generateVtgate(opts))
-	dockerComposeFile = applyInMemoryPatch(dockerComposeFile, generateVtwork(opts))
+func applyDefaultDockerPatches(dockerComposeFile []byte) []byte {
+	dockerComposeFile = applyInMemoryPatch(dockerComposeFile, generateVtctld())
+	dockerComposeFile = applyInMemoryPatch(dockerComposeFile, generateVtgate())
+	dockerComposeFile = applyInMemoryPatch(dockerComposeFile, generateVtwork())
 	return dockerComposeFile
 }
 
-func applyDockerComposePatches(
-	dockerComposeFile []byte,
-	keyspaceInfoMap map[string]keyspaceInfo,
-	externalDbInfoMap map[string]externalDbInfo,
-	vtOpts vtOptions,
-) []byte {
-	// Vtctld, vtgate, vtwork patches.
-	dockerComposeFile = applyDefaultDockerPatches(dockerComposeFile, vtOpts)
+func applyDockerComposePatches(dockerComposeFile []byte, keyspaceInfoMap map[string]keyspaceInfo, externalDbInfoMap map[string]externalDbInfo) []byte {
+	//Vtctld, vtgate, vtwork patches
+	dockerComposeFile = applyDefaultDockerPatches(dockerComposeFile)
 	for _, keyspaceData := range keyspaceInfoMap {
-		dockerComposeFile = applyKeyspaceDependentPatches(dockerComposeFile, keyspaceData, externalDbInfoMap, vtOpts)
+		dockerComposeFile = applyKeyspaceDependentPatches(dockerComposeFile, keyspaceData, externalDbInfoMap)
 	}
 
 	return dockerComposeFile
 }
 
-func applyTabletPatches(
-	dockerComposeFile []byte,
-	tabAlias int,
-	shard string,
-	keyspaceData keyspaceInfo,
-	externalDbInfoMap map[string]externalDbInfo,
-	opts vtOptions,
-) []byte {
+func applyTabletPatches(dockerComposeFile []byte, tabAlias int, shard string, keyspaceData keyspaceInfo, externalDbInfoMap map[string]externalDbInfo) []byte {
 	var dbInfo externalDbInfo
 	if val, ok := externalDbInfoMap[keyspaceData.keyspace]; ok {
 		dbInfo = val
 	}
-	dockerComposeFile = applyInMemoryPatch(dockerComposeFile, generateDefaultTablet(tabAlias+1, shard, "master", keyspaceData.keyspace, dbInfo, opts))
+	dockerComposeFile = applyInMemoryPatch(dockerComposeFile, generateDefaultTablet(strconv.Itoa(tabAlias+1), shard, "master", keyspaceData.keyspace, dbInfo))
 	for i := 0; i < keyspaceData.replicaTablets; i++ {
-		dockerComposeFile = applyInMemoryPatch(dockerComposeFile, generateDefaultTablet(tabAlias+2+i, shard, "replica", keyspaceData.keyspace, dbInfo, opts))
+		dockerComposeFile = applyInMemoryPatch(dockerComposeFile, generateDefaultTablet(strconv.Itoa(tabAlias+2+i), shard, "replica", keyspaceData.keyspace, dbInfo))
 	}
 	return dockerComposeFile
 }
 
-func generateDefaultTablet(tabAlias int, shard, role, keyspace string, dbInfo externalDbInfo, opts vtOptions) string {
+// Default Tablet
+func generateDefaultTablet(tabAlias, shard, role, keyspace string, dbInfo externalDbInfo) string {
 	externalDb := "0"
 	if dbInfo.dbName != "" {
 		externalDb = "1"
 	}
-
-	return fmt.Sprintf(`
+	data := fmt.Sprintf(`
 - op: add
-  path: /services/vttablet%[1]d
+  path: /services/vttablet%[1]s
   value:
     image: vitess/base
     ports:
-    - "15%[1]d:%[4]d"
-    - "%[5]d"
+    - "15%[1]s:%[4]s"
+    - "%[5]s"
     - "3306"
     volumes:
     - ".:/script"
     environment:
     - TOPOLOGY_FLAGS=%[7]s
-    - WEB_PORT=%[4]d
-    - GRPC_PORT=%[5]d
+    - WEB_PORT=%[4]s
+    - GRPC_PORT=%[5]s
     - CELL=%[8]s
     - KEYSPACE=%[6]s
     - SHARD=%[2]s
     - ROLE=%[3]s
-    - VTHOST=vttablet%[1]d
+    - VTHOST=vttablet%[1]s
     - EXTERNAL_DB=%[9]s
     - DB_PORT=%[10]s
     - DB_HOST=%[11]s
     - DB_USER=%[12]s
     - DB_PASS=%[13]s
     - DB_CHARSET=%[14]s
-    command: ["sh", "-c", "/script/vttablet-up.sh %[1]d"]
+    command: ["sh", "-c", "/script/vttablet-up.sh %[1]s"]
     depends_on:
       - vtctld
     healthcheck:
-        test: ["CMD-SHELL","curl localhost:%[4]d/debug/health"]
+        test: ["CMD-SHELL","curl localhost:%[4]s/debug/health"]
         interval: 30s
         timeout: 10s
         retries: 15
-`, tabAlias, shard, role, opts.webPort, opts.gRpcPort, keyspace, opts.topologyFlags, opts.cell, externalDb, dbInfo.dbPort, dbInfo.dbHost, dbInfo.dbUser, dbInfo.dbPass, dbInfo.dbCharset)
+`, tabAlias, shard, role, *webPort, *gRpcPort, keyspace, *topologyFlags, *cell, externalDb, dbInfo.dbPort, dbInfo.dbHost, dbInfo.dbUser, dbInfo.dbPass, dbInfo.dbCharset)
+
+	return data
 }
 
-func generateVtctld(opts vtOptions) string {
-	return fmt.Sprintf(`
+// Generate Vtctld
+func generateVtctld() string {
+	data := fmt.Sprintf(`
 - op: add
   path: /services/vtctld
   value:
     image: vitess/base
     ports:
-      - "15000:%[1]d"
-      - "%[2]d"
+      - "15000:%[1]s"
+      - "%[2]s"
     command: ["sh", "-c", " $$VTROOT/bin/vtctld \
         %[3]s \
         -cell %[4]s \
@@ -551,8 +472,8 @@ func generateVtctld(opts vtOptions) string {
         -backup_storage_implementation file \
         -file_backup_storage_root $$VTDATAROOT/backups \
         -logtostderr=true \
-        -port %[1]d \
-        -grpc_port %[2]d \
+        -port %[1]s \
+        -grpc_port %[2]s \
         -pid_file $$VTDATAROOT/tmp/vtctld.pid
         "]
     volumes:
@@ -561,25 +482,28 @@ func generateVtctld(opts vtOptions) string {
       - consul1
       - consul2
       - consul3
-`, opts.webPort, opts.gRpcPort, opts.topologyFlags, opts.cell)
+`, *webPort, *gRpcPort, *topologyFlags, *cell)
+
+	return data
 }
 
-func generateVtgate(opts vtOptions) string {
-	return fmt.Sprintf(`
+// Generate Vtgate
+func generateVtgate() string {
+	data := fmt.Sprintf(`
 - op: add
   path: /services/vtgate
   value:
     image: vitess/base
     ports:
-      - "15099:%[1]d"
-      - "%[2]d"
-      - "15306:%[3]d"
+      - "15099:%[1]s"
+      - "%[2]s"
+      - "15306:%[3]s"
     command: ["sh", "-c", "/script/run-forever.sh $$VTROOT/bin/vtgate \
         %[4]s \
         -logtostderr=true \
-        -port %[1]d \
-        -grpc_port %[2]d \
-        -mysql_server_port %[3]d \
+        -port %[1]s \
+        -grpc_port %[2]s \
+        -mysql_server_port %[3]s \
         -mysql_auth_server_impl none \
         -cell %[5]s \
         -cells_to_watch %[5]s \
@@ -593,40 +517,38 @@ func generateVtgate(opts vtOptions) string {
       - .:/script
     depends_on:
       - vtctld
-`, opts.webPort, opts.gRpcPort, opts.mySqlPort, opts.topologyFlags, opts.cell)
+`, *webPort, *gRpcPort, *mySqlPort, *topologyFlags, *cell)
+
+	return data
 }
 
-func generateVtwork(opts vtOptions) string {
-	return fmt.Sprintf(`
+func generateVtwork() string {
+	data := fmt.Sprintf(`
 - op: add
   path: /services/vtwork
   value:
     image: vitess/base
     ports:
-      - "15100:%[1]d"
-      - "%[2]d"
+      - "15100:%[1]s"
+      - "%[2]s"
     command: ["sh", "-c", "$$VTROOT/bin/vtworker \
         %[3]s \
         -cell %[4]s \
         -logtostderr=true \
         -service_map 'grpc-vtworker' \
-        -port %[1]d \
-        -grpc_port %[2]d \
+        -port %[1]s \
+        -grpc_port %[2]s \
         -use_v3_resharding_mode=true \
         -pid_file $$VTDATAROOT/tmp/vtwork.pid \
         "]
     depends_on:
       - vtctld
-`, opts.webPort, opts.gRpcPort, opts.topologyFlags, opts.cell)
+`, *webPort, *gRpcPort, *topologyFlags, *cell)
+
+	return data
 }
 
-func generateSchemaload(
-	tabletAliases []string,
-	postLoadFile string,
-	keyspace string,
-	dbInfo externalDbInfo,
-	opts vtOptions,
-) string {
+func generateSchemaload(tabletAliases []string, postLoadFile string, keyspace string, dbInfo externalDbInfo) string {
 	targetTab := tabletAliases[0]
 	schemaFileName := fmt.Sprintf("%s_schema_file.sql", keyspace)
 	externalDb := "0"
@@ -638,11 +560,13 @@ func generateSchemaload(
 
 	// Formatting for list in yaml
 	for i, tabletId := range tabletAliases {
+		//tabletAliases[i] = "\"vttablet" + tabletId + "\""
 		tabletAliases[i] = "vttablet" + tabletId + ": " + "{condition : service_healthy}"
 	}
+	//dependsOn := "[" + strings.Join(tabletAliases, ", ") + "]"
 	dependsOn := "depends_on: {" + strings.Join(tabletAliases, ", ") + "}"
 
-	return fmt.Sprintf(`
+	data := fmt.Sprintf(`
 - op: add
   path: /services/schemaload_%[7]s
   value:
@@ -651,8 +575,8 @@ func generateSchemaload(
       - ".:/script"
     environment:
       - TOPOLOGY_FLAGS=%[3]s
-      - WEB_PORT=%[4]d
-      - GRPC_PORT=%[5]d
+      - WEB_PORT=%[4]s
+      - GRPC_PORT=%[5]s
       - CELL=%[6]s
       - KEYSPACE=%[7]s
       - TARGETTAB=%[6]s-0000000%[2]s
@@ -663,11 +587,13 @@ func generateSchemaload(
       - EXTERNAL_DB=%[10]s
     command: ["sh", "-c", "/script/schemaload.sh"]
     %[1]s
-`, dependsOn, targetTab, opts.topologyFlags, opts.webPort, opts.gRpcPort, opts.cell, keyspace, postLoadFile, schemaFileName, externalDb)
+`, dependsOn, targetTab, *topologyFlags, *webPort, *gRpcPort, *cell, keyspace, postLoadFile, schemaFileName, externalDb)
+
+	return data
 }
 
-func generatePrimaryVIndex(tableName, column, name string) string {
-	return fmt.Sprintf(`
+func generatePrimaryVIndex(tableName, column string, name string) string {
+	data := fmt.Sprintf(`
 [{"op": "add",
 "path": "/tables/%[1]s",
 "value": 
@@ -679,10 +605,12 @@ func generatePrimaryVIndex(tableName, column, name string) string {
   ]}
 }]
 `, tableName, column, name)
+
+	return data
 }
 
 func generateVschemaLookupHash(tableName, tableKeyspace, from, to, owner string) string {
-	return fmt.Sprintf(`
+	data := fmt.Sprintf(`
 [{"op": "add",
 "path": "/vindexes/%[1]s",
 "value":
@@ -697,10 +625,12 @@ func generateVschemaLookupHash(tableName, tableKeyspace, from, to, owner string)
   }
 }]
 `, tableName, tableKeyspace, from, to, owner)
+
+	return data
 }
 
 func addToColumnVIndexes(tableName, column, referenceName string) string {
-	return fmt.Sprintf(`
+	data := fmt.Sprintf(`
 [{"op": "add",
 "path": "/tables/%[1]s/column_vindexes/-",
 "value":
@@ -710,4 +640,6 @@ func addToColumnVIndexes(tableName, column, referenceName string) string {
     }
 }]
 `, tableName, column, referenceName)
+
+	return data
 }
diff --git a/examples/local/scripts/k3s-up.sh b/examples/local/scripts/k3s-up.sh
index eb8530f15d..3b77600018 100755
--- a/examples/local/scripts/k3s-up.sh
+++ b/examples/local/scripts/k3s-up.sh
@@ -25,17 +25,6 @@ script_root=$(dirname "${BASH_SOURCE[0]}")
 # shellcheck disable=SC1091
 source ./env.sh
 
-case $(uname) in
-  Linux) ;;
-  *) echo "WARNING: unsupported platform. K3s only supports running on Linux, the k8s topology is available for local examples."; exit 1;;
-esac
-
-case $(uname -m) in
-    aarch64) ;;
-    x86_64) ;;
-    *) echo "ERROR: unsupported architecture, the k8s topology is not available for local examples."; exit 1;;
-esac
-
 k3s server --disable-agent --data-dir "${VTDATAROOT}/k3s/" --https-listen-port "${K8S_PORT}" --write-kubeconfig "${K8S_KUBECONFIG}" > "${VTDATAROOT}"/tmp/k3s.out 2>&1 &
 PID=$!
 echo $PID > "${VTDATAROOT}/tmp/k3s.pid"
diff --git a/go/stats/counters.go b/go/stats/counters.go
index f144c0ce3d..46e9b8f904 100644
--- a/go/stats/counters.go
+++ b/go/stats/counters.go
@@ -137,6 +137,9 @@ func (c *CountersWithSingleLabel) Label() string {
 
 // Add adds a value to a named counter.
 func (c *CountersWithSingleLabel) Add(name string, value int64) {
+	if value < 0 {
+		logCounterNegative.Warningf("Adding a negative value to a counter, %v should be a gauge instead", c)
+	}
 	if c.labelCombined {
 		name = StatsAllStr
 	}
@@ -196,6 +199,9 @@ func (mc *CountersWithMultiLabels) Add(names []string, value int64) {
 	if len(names) != len(mc.labels) {
 		panic("CountersWithMultiLabels: wrong number of values in Add")
 	}
+	if value < 0 {
+		logCounterNegative.Warningf("Adding a negative value to a counter, %v should be a gauge instead", mc)
+	}
 	mc.counters.add(safeJoinLabels(names, mc.combinedLabels), value)
 }
 
diff --git a/go/vt/binlog/slave_connection.go b/go/vt/binlog/slave_connection.go
index c83b70c368..64367757b7 100644
--- a/go/vt/binlog/slave_connection.go
+++ b/go/vt/binlog/slave_connection.go
@@ -17,15 +17,13 @@ limitations under the License.
 package binlog
 
 import (
-	crand "crypto/rand"
 	"fmt"
-	"math"
-	"math/big"
 	"sync"
 
 	"golang.org/x/net/context"
 
 	"vitess.io/vitess/go/mysql"
+	"vitess.io/vitess/go/pools"
 	"vitess.io/vitess/go/vt/dbconfigs"
 	"vitess.io/vitess/go/vt/log"
 )
@@ -48,16 +46,13 @@ type SlaveConnection struct {
 	wg      sync.WaitGroup
 }
 
-// We use a random slaveid deprecating IDPool which was causing problems with RDS
-// either because it was using ids in the same range that we generate or because
-// the pool reuses ids.
-func getSlaveID() uint32 {
-	max := big.NewInt(math.MaxInt32)
-	id, _ := crand.Int(crand.Reader, max)
-	return uint32(id.Int64())
-}
-
 // NewSlaveConnection creates a new slave connection to the mysqld instance.
+// It uses a pools.IDPool to ensure that the server IDs used to connect are
+// unique within this process. This is done with the assumptions that:
+//
+// 1) No other processes are making fake slave connections to our mysqld.
+// 2) No real slave servers will have IDs in the range 1-N where N is the peak
+//    number of concurrent fake slave connections we will ever make.
 func NewSlaveConnection(cp dbconfigs.Connector) (*SlaveConnection, error) {
 	conn, err := connectForReplication(cp)
 	if err != nil {
@@ -67,7 +62,7 @@ func NewSlaveConnection(cp dbconfigs.Connector) (*SlaveConnection, error) {
 	sc := &SlaveConnection{
 		Conn:    conn,
 		cp:      cp,
-		slaveID: getSlaveID(),
+		slaveID: slaveIDPool.Get(),
 	}
 	log.Infof("new slave connection: slaveID=%d", sc.slaveID)
 	return sc, nil
@@ -89,6 +84,9 @@ func connectForReplication(cp dbconfigs.Connector) (*mysql.Conn, error) {
 	return conn, nil
 }
 
+// slaveIDPool is the IDPool for server IDs used to connect as a slave.
+var slaveIDPool = pools.NewIDPool()
+
 // StartBinlogDumpFromCurrent requests a replication binlog dump from
 // the current position.
 func (sc *SlaveConnection) StartBinlogDumpFromCurrent(ctx context.Context) (mysql.Position, <-chan mysql.BinlogEvent, error) {
@@ -282,7 +280,8 @@ func (sc *SlaveConnection) Close() {
 			sc.cancel = nil
 		}
 
-		log.Infof("closing slave MySQL client with slaveID %v", sc.slaveID)
+		log.Infof("closing slave MySQL client, recycling slaveID %v", sc.slaveID)
 		sc.Conn = nil
+		slaveIDPool.Put(sc.slaveID)
 	}
 }
diff --git a/go/vt/servenv/servenv.go b/go/vt/servenv/servenv.go
index 5435a67e17..de3165a723 100644
--- a/go/vt/servenv/servenv.go
+++ b/go/vt/servenv/servenv.go
@@ -206,8 +206,8 @@ func RunDefault() {
 func ParseFlags(cmd string) {
 	flag.Parse()
 
-	AppVersion.Print()
 	if *Version {
+		AppVersion.Print()
 		os.Exit(0)
 	}
 
diff --git a/go/vt/sqlparser/analyzer.go b/go/vt/sqlparser/analyzer.go
index c221787fa4..113a887fb3 100644
--- a/go/vt/sqlparser/analyzer.go
+++ b/go/vt/sqlparser/analyzer.go
@@ -163,16 +163,6 @@ func IsDML(sql string) bool {
 	return false
 }
 
-//IsDMLStatement returns true if the query is an INSERT, UPDATE or DELETE statement.
-func IsDMLStatement(stmt Statement) bool {
-	switch stmt.(type) {
-	case *Insert, *Update, *Delete:
-		return true
-	}
-
-	return false
-}
-
 // SplitAndExpression breaks up the Expr into AND-separated conditions
 // and appends them to filters. Outer parenthesis are removed. Precedence
 // should be taken into account if expressions are recombined.
diff --git a/go/vt/vtexplain/vtexplain_vtgate.go b/go/vt/vtexplain/vtexplain_vtgate.go
index 95259873f6..9cedd20807 100644
--- a/go/vt/vtexplain/vtexplain_vtgate.go
+++ b/go/vt/vtexplain/vtexplain_vtgate.go
@@ -72,7 +72,7 @@ func initVtgateExecutor(vSchemaStr string, opts *Options) error {
 
 func newFakeResolver(opts *Options, hc discovery.HealthCheck, serv srvtopo.Server, cell string) *vtgate.Resolver {
 	ctx := context.Background()
-	gw := vtgate.GatewayCreator()(ctx, hc, serv, cell, 3)
+	gw := vtgate.NewTabletGateway(ctx, hc, serv, cell, 3)
 	gw.WaitForTablets(ctx, []topodatapb.TabletType{topodatapb.TabletType_REPLICA})
 
 	txMode := vtgatepb.TransactionMode_MULTI
diff --git a/go/vt/vtgate/engine/delete.go b/go/vt/vtgate/engine/delete.go
index 409ce0d5d9..c1bf66db08 100644
--- a/go/vt/vtgate/engine/delete.go
+++ b/go/vt/vtgate/engine/delete.go
@@ -166,7 +166,7 @@ func (del *Delete) execDeleteEqual(vcursor VCursor, bindVars map[string]*querypb
 			return nil, vterrors.Wrap(err, "execDeleteEqual")
 		}
 	}
-	return execShard(vcursor, del.Query, bindVars, rs, true /* rollbackOnError */, true /* canAutocommit */)
+	return execShard(vcursor, del.Query, bindVars, rs, true /* isDML */, true /* canAutocommit */)
 }
 
 // deleteVindexEntries performs an delete if table owns vindex.
@@ -231,6 +231,6 @@ func (del *Delete) execDeleteByDestination(vcursor VCursor, bindVars map[string]
 		}
 	}
 	autocommit := (len(rss) == 1 || del.MultiShardAutocommit) && vcursor.AutocommitApproval()
-	res, errs := vcursor.ExecuteMultiShard(rss, queries, true /* rollbackOnError */, autocommit)
+	res, errs := vcursor.ExecuteMultiShard(rss, queries, true /* isDML */, autocommit)
 	return res, vterrors.Aggregate(errs)
 }
diff --git a/go/vt/vtgate/engine/insert.go b/go/vt/vtgate/engine/insert.go
index e08d5398ea..625a465623 100644
--- a/go/vt/vtgate/engine/insert.go
+++ b/go/vt/vtgate/engine/insert.go
@@ -277,7 +277,7 @@ func (ins *Insert) execInsertSharded(vcursor VCursor, bindVars map[string]*query
 	}
 
 	autocommit := (len(rss) == 1 || ins.MultiShardAutocommit) && vcursor.AutocommitApproval()
-	result, errs := vcursor.ExecuteMultiShard(rss, queries, true /* rollbackOnError */, autocommit)
+	result, errs := vcursor.ExecuteMultiShard(rss, queries, true /* isDML */, autocommit)
 	if errs != nil {
 		return nil, vterrors.Wrap(vterrors.Aggregate(errs), "execInsertSharded")
 	}
diff --git a/go/vt/vtgate/engine/primitive.go b/go/vt/vtgate/engine/primitive.go
index 1ee2389b04..34a8d96a22 100644
--- a/go/vt/vtgate/engine/primitive.go
+++ b/go/vt/vtgate/engine/primitive.go
@@ -57,16 +57,16 @@ type VCursor interface {
 	RecordWarning(warning *querypb.QueryWarning)
 
 	// V3 functions.
-	Execute(method string, query string, bindvars map[string]*querypb.BindVariable, rollbackOnError bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error)
+	Execute(method string, query string, bindvars map[string]*querypb.BindVariable, isDML bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error)
 	AutocommitApproval() bool
 
 	// Shard-level functions.
-	ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, rollbackOnError, canAutocommit bool) (*sqltypes.Result, []error)
+	ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, isDML, canAutocommit bool) (*sqltypes.Result, []error)
 	ExecuteStandalone(query string, bindvars map[string]*querypb.BindVariable, rs *srvtopo.ResolvedShard) (*sqltypes.Result, error)
 	StreamExecuteMulti(query string, rss []*srvtopo.ResolvedShard, bindVars []map[string]*querypb.BindVariable, callback func(reply *sqltypes.Result) error) error
 
 	// Keyspace ID level functions.
-	ExecuteKeyspaceID(keyspace string, ksid []byte, query string, bindVars map[string]*querypb.BindVariable, rollbackOnError, autocommit bool) (*sqltypes.Result, error)
+	ExecuteKeyspaceID(keyspace string, ksid []byte, query string, bindVars map[string]*querypb.BindVariable, isDML, autocommit bool) (*sqltypes.Result, error)
 
 	// Resolver methods, from key.Destination to srvtopo.ResolvedShard.
 	// Will replace all of the Topo functions.
diff --git a/go/vt/vtgate/engine/route.go b/go/vt/vtgate/engine/route.go
index d1d78a9bc1..8a0f2f326a 100644
--- a/go/vt/vtgate/engine/route.go
+++ b/go/vt/vtgate/engine/route.go
@@ -267,7 +267,7 @@ func (route *Route) execute(vcursor VCursor, bindVars map[string]*querypb.BindVa
 	}
 
 	queries := getQueries(route.Query, bvs)
-	result, errs := vcursor.ExecuteMultiShard(rss, queries, false /* rollbackOnError */, false /* autocommit */)
+	result, errs := vcursor.ExecuteMultiShard(rss, queries, false /* isDML */, false /* autocommit */)
 
 	if errs != nil {
 		if route.ScatterErrorsAsWarnings {
@@ -362,7 +362,7 @@ func (route *Route) GetFields(vcursor VCursor, bindVars map[string]*querypb.Bind
 		// This code is unreachable. It's just a sanity check.
 		return nil, fmt.Errorf("no shards for keyspace: %s", route.Keyspace.Name)
 	}
-	qr, err := execShard(vcursor, route.FieldQuery, bindVars, rss[0], false /* rollbackOnError */, false /* canAutocommit */)
+	qr, err := execShard(vcursor, route.FieldQuery, bindVars, rss[0], false /* isDML */, false /* canAutocommit */)
 	if err != nil {
 		return nil, err
 	}
@@ -518,14 +518,14 @@ func resolveKeyspaceID(vcursor VCursor, vindex vindexes.SingleColumn, vindexKey
 	}
 }
 
-func execShard(vcursor VCursor, query string, bindVars map[string]*querypb.BindVariable, rs *srvtopo.ResolvedShard, rollbackOnError, canAutocommit bool) (*sqltypes.Result, error) {
+func execShard(vcursor VCursor, query string, bindVars map[string]*querypb.BindVariable, rs *srvtopo.ResolvedShard, isDML, canAutocommit bool) (*sqltypes.Result, error) {
 	autocommit := canAutocommit && vcursor.AutocommitApproval()
 	result, errs := vcursor.ExecuteMultiShard([]*srvtopo.ResolvedShard{rs}, []*querypb.BoundQuery{
 		{
 			Sql:           query,
 			BindVariables: bindVars,
 		},
-	}, rollbackOnError, autocommit)
+	}, isDML, autocommit)
 	return result, vterrors.Aggregate(errs)
 }
 
diff --git a/go/vt/vtgate/engine/send.go b/go/vt/vtgate/engine/send.go
deleted file mode 100644
index 8a8d0a880f..0000000000
--- a/go/vt/vtgate/engine/send.go
+++ /dev/null
@@ -1,116 +0,0 @@
-package engine
-
-import (
-	"encoding/json"
-
-	"vitess.io/vitess/go/sqltypes"
-	"vitess.io/vitess/go/vt/key"
-	"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/vindexes"
-
-	querypb "vitess.io/vitess/go/vt/proto/query"
-)
-
-var _ Primitive = (*Send)(nil)
-
-// Send is an operator to send query to the specific keyspace, tabletType and destination
-type Send struct {
-	// Keyspace specifies the keyspace to send the query to.
-	Keyspace *vindexes.Keyspace
-
-	// TargetDestination specifies an explicit target destination to send the query to.
-	// This bypases the core of the v3 engine.
-	TargetDestination key.Destination
-
-	// Query specifies the query to be executed.
-	Query string
-
-	// NoAutoCommit specifies if we need to check autocommit behaviour
-	NoAutoCommit bool
-
-	noInputs
-}
-
-// MarshalJSON serializes the Send into a JSON representation.
-// It's used for testing and diagnostics.
-func (s *Send) MarshalJSON() ([]byte, error) {
-	marshalSend := struct {
-		Opcode            string
-		Keyspace          *vindexes.Keyspace
-		TargetDestination key.Destination
-		Query             string
-		NoAutoCommit      bool
-	}{
-		Opcode:            "Send",
-		Keyspace:          s.Keyspace,
-		TargetDestination: s.TargetDestination,
-		NoAutoCommit:      s.NoAutoCommit,
-		Query:             s.Query,
-	}
-
-	return json.Marshal(marshalSend)
-}
-
-// RouteType implements Primitive interface
-func (s *Send) RouteType() string {
-	if s.NoAutoCommit {
-		return "SendNoAutoCommit"
-	}
-
-	return "Send"
-}
-
-// GetKeyspaceName implements Primitive interface
-func (s *Send) GetKeyspaceName() string {
-	return s.Keyspace.Name
-}
-
-// GetTableName implements Primitive interface
-func (s *Send) GetTableName() string {
-	return ""
-}
-
-// Execute implements Primitive interface
-func (s *Send) Execute(vcursor VCursor, bindVars map[string]*query.BindVariable, _ bool) (*sqltypes.Result, error) {
-	rss, _, err := vcursor.ResolveDestinations(s.Keyspace.Name, nil, []key.Destination{s.TargetDestination})
-	if err != nil {
-		return nil, vterrors.Wrap(err, "sendExecute")
-	}
-
-	if !s.Keyspace.Sharded && len(rss) != 1 {
-		return nil, vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "Keyspace does not have exactly one shard: %v", rss)
-	}
-
-	queries := make([]*querypb.BoundQuery, len(rss))
-	for i := range rss {
-		queries[i] = &querypb.BoundQuery{
-			Sql:           s.Query,
-			BindVariables: bindVars,
-		}
-	}
-
-	canAutocommit := false
-	if !s.NoAutoCommit {
-		canAutocommit = len(rss) == 1 && vcursor.AutocommitApproval()
-	}
-
-	rollbackOnError := !s.NoAutoCommit // for non-dml queries, there's no need to do a rollback
-	result, errs := vcursor.ExecuteMultiShard(rss, queries, rollbackOnError, canAutocommit)
-	err = vterrors.Aggregate(errs)
-	if err != nil {
-		return nil, err
-	}
-	return result, nil
-}
-
-// StreamExecute implements Primitive interface
-func (s *Send) StreamExecute(vcursor VCursor, bindVars map[string]*query.BindVariable, wantields bool, callback func(*sqltypes.Result) error) error {
-	return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "not reachable") // TODO: systay - this should work
-}
-
-// GetFields implements Primitive interface
-func (s *Send) GetFields(vcursor VCursor, bindVars map[string]*query.BindVariable) (*sqltypes.Result, error) {
-	return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "not reachable")
-}
diff --git a/go/vt/vtgate/engine/update.go b/go/vt/vtgate/engine/update.go
index 684fe65d7d..565f86e1e1 100644
--- a/go/vt/vtgate/engine/update.go
+++ b/go/vt/vtgate/engine/update.go
@@ -173,7 +173,7 @@ func (upd *Update) execUpdateEqual(vcursor VCursor, bindVars map[string]*querypb
 			return nil, vterrors.Wrap(err, "execUpdateEqual")
 		}
 	}
-	return execShard(vcursor, upd.Query, bindVars, rs, true /* rollbackOnError */, true /* canAutocommit */)
+	return execShard(vcursor, upd.Query, bindVars, rs, true /* isDML */, true /* canAutocommit */)
 }
 
 // updateVindexEntries performs an update when a vindex is being modified
@@ -260,6 +260,6 @@ func (upd *Update) execUpdateByDestination(vcursor VCursor, bindVars map[string]
 	}
 
 	autocommit := (len(rss) == 1 || upd.MultiShardAutocommit) && vcursor.AutocommitApproval()
-	result, errs := vcursor.ExecuteMultiShard(rss, queries, true /* rollbackOnError */, autocommit)
+	result, errs := vcursor.ExecuteMultiShard(rss, queries, true /* isDML */, autocommit)
 	return result, vterrors.Aggregate(errs)
 }
diff --git a/go/vt/vtgate/executor.go b/go/vt/vtgate/executor.go
index 3535022cd9..49f4efea8e 100644
--- a/go/vt/vtgate/executor.go
+++ b/go/vt/vtgate/executor.go
@@ -173,7 +173,7 @@ func (e *Executor) execute(ctx context.Context, safeSession *SafeSession, sql st
 	if err != nil {
 		return nil, err
 	}
-	// ks@replica , :-80-> ks:-80, -80@replica, ks:-80@replica
+
 	if safeSession.InTransaction() && destTabletType != topodatapb.TabletType_MASTER {
 		return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "transactions are supported only for master tablet types, current type: %v", destTabletType)
 	}
@@ -207,7 +207,7 @@ func (e *Executor) execute(ctx context.Context, safeSession *SafeSession, sql st
 
 	switch specStmt := stmt.(type) {
 	case *sqlparser.Select, *sqlparser.Union:
-		return e.handleExec(ctx, safeSession, sql, bindVars, logStats, stmtType)
+		return e.handleExec(ctx, safeSession, sql, bindVars, destKeyspace, destTabletType, dest, logStats, stmtType)
 	case *sqlparser.Insert, *sqlparser.Update, *sqlparser.Delete:
 		safeSession := safeSession
 
@@ -232,7 +232,7 @@ func (e *Executor) execute(ctx context.Context, safeSession *SafeSession, sql st
 		// at the beginning, but never after.
 		safeSession.SetAutocommittable(mustCommit)
 
-		qr, err := e.handleExec(ctx, safeSession, sql, bindVars, logStats, stmtType)
+		qr, err := e.handleExec(ctx, safeSession, sql, bindVars, destKeyspace, destTabletType, dest, logStats, stmtType)
 		if err != nil {
 			return nil, err
 		}
@@ -265,9 +265,54 @@ func (e *Executor) execute(ctx context.Context, safeSession *SafeSession, sql st
 	return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "unrecognized statement: %s", sql)
 }
 
-func (e *Executor) handleExec(ctx context.Context, safeSession *SafeSession, sql string, bindVars map[string]*querypb.BindVariable, logStats *LogStats, stmtType sqlparser.StatementType) (*sqltypes.Result, error) {
+func (e *Executor) handleExec(ctx context.Context, safeSession *SafeSession, sql string, bindVars map[string]*querypb.BindVariable, destKeyspace string, destTabletType topodatapb.TabletType, dest key.Destination, logStats *LogStats, stmtType sqlparser.StatementType) (*sqltypes.Result, error) {
+	if dest != nil {
+		if destKeyspace == "" {
+			return nil, errNoKeyspace
+		}
+
+		switch dest.(type) {
+		case key.DestinationExactKeyRange:
+			stmtType := sqlparser.Preview(sql)
+			if stmtType == sqlparser.StmtInsert {
+				return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "range queries not supported for inserts: %s", safeSession.TargetString)
+			}
+
+		}
+
+		execStart := time.Now()
+		if e.normalize {
+			query, comments := sqlparser.SplitMarginComments(sql)
+			stmt, err := sqlparser.Parse(query)
+			if err != nil {
+				return nil, err
+			}
+			rewriteResult, err := sqlparser.PrepareAST(stmt, bindVars, "vtg")
+			if err != nil {
+				return nil, err
+			}
+			normalized := sqlparser.String(rewriteResult.AST)
+			sql = comments.Leading + normalized + comments.Trailing
+			neededBindVariables, err := e.createNeededBindVariables(rewriteResult.BindVarNeeds, safeSession)
+			if err != nil {
+				return nil, err
+			}
+			for k, v := range neededBindVariables {
+				bindVars[k] = v
+			}
+		}
+		logStats.PlanTime = execStart.Sub(logStats.StartTime)
+		logStats.SQL = sql
+		logStats.BindVariables = bindVars
+		result, err := e.resolver.Execute(ctx, sql, bindVars, destKeyspace, destTabletType, dest, safeSession, false /* notInTransaction */, safeSession.Options, logStats, true /* canAutocommit */)
+		logStats.ExecuteTime = time.Since(execStart)
+		e.updateQueryCounts("ShardDirect", "", "", int64(logStats.ShardQueries))
+		return result, err
+	}
+
+	// V3 mode.
 	query, comments := sqlparser.SplitMarginComments(sql)
-	vcursor, _ := newVCursorImpl(ctx, safeSession, comments, e, logStats, e.VSchema(), e.resolver.resolver)
+	vcursor := newVCursorImpl(ctx, safeSession, destKeyspace, destTabletType, comments, e, logStats, e.VSchema(), e.resolver.resolver)
 	plan, err := e.getPlan(
 		vcursor,
 		query,
@@ -310,7 +355,7 @@ func (e *Executor) handleExec(ctx context.Context, safeSession *SafeSession, sql
 	}
 
 	// Check if there was partial DML execution. If so, rollback the transaction.
-	if err != nil && safeSession.InTransaction() && vcursor.rollbackOnPartialExec {
+	if err != nil && safeSession.InTransaction() && vcursor.hasPartialDML {
 		_ = e.txConn.Rollback(ctx, safeSession)
 		err = vterrors.Errorf(vtrpcpb.Code_ABORTED, "transaction rolled back due to partial DML execution: %v", err)
 	}
@@ -1172,7 +1217,7 @@ func (e *Executor) StreamExecute(ctx context.Context, method string, safeSession
 		bindVars = make(map[string]*querypb.BindVariable)
 	}
 	query, comments := sqlparser.SplitMarginComments(sql)
-	vcursor, _ := newVCursorImpl(ctx, safeSession, comments, e, logStats, e.VSchema(), e.resolver.resolver)
+	vcursor := newVCursorImpl(ctx, safeSession, target.Keyspace, target.TabletType, comments, e, logStats, e.VSchema(), e.resolver.resolver)
 
 	// check if this is a stream statement for messaging
 	// TODO: support keyRange syntax
@@ -1331,7 +1376,8 @@ func (e *Executor) getPlan(vcursor *vcursorImpl, sql string, comments sqlparser.
 	if e.VSchema() == nil {
 		return nil, errors.New("vschema not initialized")
 	}
-	planKey := vcursor.planPrefixKey() + ":" + sql
+	keyspace := vcursor.keyspace
+	planKey := keyspace + vindexes.TabletTypeSuffix[vcursor.tabletType] + ":" + sql
 	if plan, ok := e.plans.Get(planKey); ok {
 		return plan.(*engine.Plan), nil
 	}
@@ -1363,7 +1409,7 @@ func (e *Executor) getPlan(vcursor *vcursorImpl, sql string, comments sqlparser.
 		logStats.BindVariables = bindVars
 	}
 
-	planKey = vcursor.planPrefixKey() + ":" + normalized
+	planKey = keyspace + vindexes.TabletTypeSuffix[vcursor.tabletType] + ":" + normalized
 	if plan, ok := e.plans.Get(planKey); ok {
 		return plan.(*engine.Plan), nil
 	}
@@ -1621,7 +1667,7 @@ func (e *Executor) prepare(ctx context.Context, safeSession *SafeSession, sql st
 func (e *Executor) handlePrepare(ctx context.Context, safeSession *SafeSession, sql string, bindVars map[string]*querypb.BindVariable, destKeyspace string, destTabletType topodatapb.TabletType, logStats *LogStats) ([]*querypb.Field, error) {
 	// V3 mode.
 	query, comments := sqlparser.SplitMarginComments(sql)
-	vcursor, _ := newVCursorImpl(ctx, safeSession, comments, e, logStats, e.VSchema(), e.resolver.resolver)
+	vcursor := newVCursorImpl(ctx, safeSession, destKeyspace, destTabletType, comments, e, logStats, e.VSchema(), e.resolver.resolver)
 	plan, err := e.getPlan(
 		vcursor,
 		query,
diff --git a/go/vt/vtgate/gateway.go b/go/vt/vtgate/gateway.go
deleted file mode 100644
index 8bf3083337..0000000000
--- a/go/vt/vtgate/gateway.go
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
-Copyright 2019 The Vitess Authors.
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-    http://www.apache.org/licenses/LICENSE-2.0
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package vtgate
-
-import (
-	"flag"
-	"time"
-
-	"golang.org/x/net/context"
-	"vitess.io/vitess/go/flagutil"
-	"vitess.io/vitess/go/vt/log"
-
-	"vitess.io/vitess/go/vt/discovery"
-	"vitess.io/vitess/go/vt/srvtopo"
-	"vitess.io/vitess/go/vt/vttablet/queryservice"
-
-	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
-)
-
-// This file contains the Gateway interface definition, and the
-// implementations registry.
-// A Gateway can take
-// a query targeted to a keyspace/shard/tablet_type and send it off.
-
-var (
-	implementation       = flag.String("gateway_implementation", "discoverygateway", "The implementation of gateway")
-	initialTabletTimeout = flag.Duration("gateway_initial_tablet_timeout", 30*time.Second, "At startup, the gateway will wait up to that duration to get one tablet per keyspace/shard/tablettype")
-
-	// KeyspacesToWatch - if provided this specifies which keyspaces should be
-	// visible to a vtgate. By default the vtgate will allow access to any
-	// keyspace.
-	KeyspacesToWatch flagutil.StringListValue
-)
-
-func init() {
-	flag.Var(&KeyspacesToWatch, "keyspaces_to_watch", "Specifies which keyspaces this vtgate should have access to while routing queries or accessing the vschema")
-}
-
-// A Gateway is the query processing module for each shard,
-// which is used by ScatterConn.
-type Gateway interface {
-	// the query service that this Gateway wraps around
-	queryservice.QueryService
-
-	// WaitForTablets asks the gateway to wait for the provided
-	// tablets types to be available. It the context is canceled
-	// before the end, it should return ctx.Err().
-	// The error returned will have specific effects:
-	// - nil: keep going with startup.
-	// - context.DeadlineExceeded: log a warning that we didn't get
-	//   all tablets, and keep going with startup.
-	// - any other error: log.Fatalf out.
-	WaitForTablets(ctx context.Context, tabletTypesToWait []topodatapb.TabletType) error
-
-	// RegisterStats registers exported stats for the gateway
-	RegisterStats()
-
-	// CacheStatus returns a list of TabletCacheStatus per shard / tablet type.
-	CacheStatus() TabletCacheStatusList
-}
-
-// Creator is the factory method which can create the actual gateway object.
-type Creator func(ctx context.Context, hc discovery.HealthCheck, serv srvtopo.Server, cell string, retryCount int) Gateway
-
-var creators = make(map[string]Creator)
-
-// RegisterGatewayCreator registers a Creator with given name.
-func RegisterGatewayCreator(name string, gc Creator) {
-	if _, ok := creators[name]; ok {
-		log.Fatalf("Gateway %s already exists", name)
-	}
-	creators[name] = gc
-}
-
-// GatewayCreator returns the Creator specified by the gateway_implementation flag.
-func GatewayCreator() Creator {
-	gc, ok := creators[*implementation]
-	if !ok {
-		log.Exitf("No gateway registered as %s", *implementation)
-	}
-	return gc
-}
-
-// WaitForTablets is a helper method to wait for the provided tablets,
-// up until the *initialTabletTimeout. It will log what it is doing.
-// Note it has the same name as the Gateway's interface method, as it
-// just calls it.
-func WaitForTablets(gw Gateway, tabletTypesToWait []topodatapb.TabletType) error {
-	log.Infof("Gateway waiting for serving tablets of types %v ...", tabletTypesToWait)
-	ctx, cancel := context.WithTimeout(context.Background(), *initialTabletTimeout)
-	defer cancel()
-
-	err := gw.WaitForTablets(ctx, tabletTypesToWait)
-	switch err {
-	case nil:
-		// Log so we know everything is fine.
-		log.Infof("Waiting for tablets completed")
-	case context.DeadlineExceeded:
-		// In this scenario, we were able to reach the
-		// topology service, but some tablets may not be
-		// ready. We just warn and keep going.
-		log.Warningf("Timeout waiting for all keyspaces / shards to have healthy tablets of types %v, may be in degraded mode", tabletTypesToWait)
-		err = nil
-	default:
-		// Nothing to do here, the caller will log.Fatalf.
-	}
-	return err
-}
diff --git a/go/vt/vtgate/planbuilder/builder.go b/go/vt/vtgate/planbuilder/builder.go
index ed4df32b86..bd01633352 100644
--- a/go/vt/vtgate/planbuilder/builder.go
+++ b/go/vt/vtgate/planbuilder/builder.go
@@ -122,8 +122,6 @@ type ContextVSchema interface {
 	FindTablesOrVindex(tablename sqlparser.TableName) ([]*vindexes.Table, vindexes.Vindex, string, topodatapb.TabletType, key.Destination, error)
 	DefaultKeyspace() (*vindexes.Keyspace, error)
 	TargetString() string
-	Destination() key.Destination
-	TabletType() topodatapb.TabletType
 }
 
 //-------------------------------------------------------------------------
@@ -275,42 +273,37 @@ func Build(query string, vschema ContextVSchema) (*engine.Plan, error) {
 func BuildFromStmt(query string, stmt sqlparser.Statement, vschema ContextVSchema, bindVarNeeds sqlparser.BindVarNeeds) (*engine.Plan, error) {
 	var err error
 	var instruction engine.Primitive
-
-	if vschema.Destination() != nil {
-		instruction, err = buildPlanForBypass(stmt, vschema)
-	} else {
-		switch stmt := stmt.(type) {
-		case *sqlparser.Select:
-			instruction, err = buildSelectPlan(stmt, vschema)
-		case *sqlparser.Insert:
-			instruction, err = buildInsertPlan(stmt, vschema)
-		case *sqlparser.Update:
-			instruction, err = buildUpdatePlan(stmt, vschema)
-		case *sqlparser.Delete:
-			instruction, err = buildDeletePlan(stmt, vschema)
-		case *sqlparser.Union:
-			instruction, err = buildUnionPlan(stmt, vschema)
-		case *sqlparser.Set:
-			return nil, errors.New("unsupported construct: set")
-		case *sqlparser.Show:
-			return nil, errors.New("unsupported construct: show")
-		case *sqlparser.DDL:
-			return nil, errors.New("unsupported construct: ddl")
-		case *sqlparser.DBDDL:
-			return nil, errors.New("unsupported construct: ddl on database")
-		case *sqlparser.OtherRead:
-			return nil, errors.New("unsupported construct: other read")
-		case *sqlparser.OtherAdmin:
-			return nil, errors.New("unsupported construct: other admin")
-		case *sqlparser.Begin:
-			return nil, errors.New("unsupported construct: begin")
-		case *sqlparser.Commit:
-			return nil, errors.New("unsupported construct: commit")
-		case *sqlparser.Rollback:
-			return nil, errors.New("unsupported construct: rollback")
-		default:
-			return nil, fmt.Errorf("BUG: unexpected statement type: %T", stmt)
-		}
+	switch stmt := stmt.(type) {
+	case *sqlparser.Select:
+		instruction, err = buildSelectPlan(stmt, vschema)
+	case *sqlparser.Insert:
+		instruction, err = buildInsertPlan(stmt, vschema)
+	case *sqlparser.Update:
+		instruction, err = buildUpdatePlan(stmt, vschema)
+	case *sqlparser.Delete:
+		instruction, err = buildDeletePlan(stmt, vschema)
+	case *sqlparser.Union:
+		instruction, err = buildUnionPlan(stmt, vschema)
+	case *sqlparser.Set:
+		return nil, errors.New("unsupported construct: set")
+	case *sqlparser.Show:
+		return nil, errors.New("unsupported construct: show")
+	case *sqlparser.DDL:
+		return nil, errors.New("unsupported construct: ddl")
+	case *sqlparser.DBDDL:
+		return nil, errors.New("unsupported construct: ddl on database")
+	case *sqlparser.OtherRead:
+		return nil, errors.New("unsupported construct: other read")
+	case *sqlparser.OtherAdmin:
+		return nil, errors.New("unsupported construct: other admin")
+	case *sqlparser.Begin:
+		return nil, errors.New("unsupported construct: begin")
+	case *sqlparser.Commit:
+		return nil, errors.New("unsupported construct: commit")
+	case *sqlparser.Rollback:
+		return nil, errors.New("unsupported construct: rollback")
+	default:
+		return nil, fmt.Errorf("BUG: unexpected statement type: %T", stmt)
 	}
 	if err != nil {
 		return nil, err
diff --git a/go/vt/vtgate/planbuilder/bypass.go b/go/vt/vtgate/planbuilder/bypass.go
deleted file mode 100644
index 2ffb3f415a..0000000000
--- a/go/vt/vtgate/planbuilder/bypass.go
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
-Copyright 2019 The Vitess Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package planbuilder
-
-import (
-	"vitess.io/vitess/go/vt/key"
-	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
-	"vitess.io/vitess/go/vt/sqlparser"
-	"vitess.io/vitess/go/vt/vterrors"
-	"vitess.io/vitess/go/vt/vtgate/engine"
-)
-
-func buildPlanForBypass(stmt sqlparser.Statement, vschema ContextVSchema) (engine.Primitive, error) {
-	switch vschema.Destination().(type) {
-	case key.DestinationExactKeyRange:
-		if _, ok := stmt.(*sqlparser.Insert); ok {
-			return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "range queries not supported for inserts: %s", vschema.TargetString())
-		}
-	}
-
-	keyspace, err := vschema.DefaultKeyspace()
-	if err != nil {
-		return nil, err
-	}
-	return &engine.Send{
-		Keyspace:          keyspace,
-		TargetDestination: vschema.Destination(),
-		Query:             sqlparser.String(stmt),
-		NoAutoCommit:      !sqlparser.IsDMLStatement(stmt),
-	}, nil
-}
diff --git a/go/vt/vtgate/scatter_conn.go b/go/vt/vtgate/scatter_conn.go
index d3657aab4c..bce0a26425 100644
--- a/go/vt/vtgate/scatter_conn.go
+++ b/go/vt/vtgate/scatter_conn.go
@@ -47,7 +47,7 @@ type ScatterConn struct {
 	timings              *stats.MultiTimings
 	tabletCallErrorCount *stats.CountersWithMultiLabels
 	txConn               *TxConn
-	gateway              Gateway
+	gateway              *tabletGateway
 	healthCheck          discovery.HealthCheck
 }
 
@@ -69,7 +69,7 @@ type shardActionFunc func(rs *srvtopo.ResolvedShard, i int) error
 type shardActionTransactionFunc func(rs *srvtopo.ResolvedShard, i int, shouldBegin bool, transactionID int64) (int64, error)
 
 // NewScatterConn creates a new ScatterConn.
-func NewScatterConn(statsName string, txConn *TxConn, gw Gateway, hc discovery.HealthCheck) *ScatterConn {
+func NewScatterConn(statsName string, txConn *TxConn, gw *tabletGateway, hc discovery.HealthCheck) *ScatterConn {
 	tabletCallErrorCountStatsName := ""
 	if statsName != "" {
 		tabletCallErrorCountStatsName = statsName + "ErrorCount"
diff --git a/go/vt/vtgate/discoverygateway.go b/go/vt/vtgate/tabletgateway.go
similarity index 80%
rename from go/vt/vtgate/discoverygateway.go
rename to go/vt/vtgate/tabletgateway.go
index 474ec99d35..1e7db1a6f3 100644
--- a/go/vt/vtgate/discoverygateway.go
+++ b/go/vt/vtgate/tabletgateway.go
@@ -19,13 +19,12 @@ package vtgate
 import (
 	"flag"
 	"fmt"
+	"golang.org/x/net/context"
 	"math/rand"
 	"sort"
 	"strings"
 	"sync"
 	"time"
-
-	"golang.org/x/net/context"
 	"vitess.io/vitess/go/vt/topotools"
 
 	"vitess.io/vitess/go/flagutil"
@@ -45,6 +44,11 @@ import (
 )
 
 var (
+	_                    = flag.String("gateway_implementation", "", "Deprecated")
+	initialTabletTimeout = flag.Duration("gateway_initial_tablet_timeout", 30*time.Second, "At startup, the gateway will wait up to that duration to get one tablet per keyspace/shard/tablettype")
+	// KeyspacesToWatch - if provided this specifies which keyspaces should be
+	// visible to a vtgate. By default the vtgate will allow access to any
+	// keyspace.
 	cellsToWatch        = flag.String("cells_to_watch", "", "comma-separated list of cells for watching tablets")
 	refreshInterval     = flag.Duration("tablet_refresh_interval", 1*time.Minute, "tablet refresh interval")
 	refreshKnownTablets = flag.Bool("tablet_refresh_known_tablets", true, "tablet refresh reloads the tablet address/port map from topo in case it changes")
@@ -52,20 +56,17 @@ var (
 
 	allowedTabletTypes []topodatapb.TabletType
 
-	tabletFilters flagutil.StringListValue
-)
-
-const (
-	gatewayImplementationDiscovery = "discoverygateway"
+	tabletFilters    flagutil.StringListValue
+	KeyspacesToWatch flagutil.StringListValue
 )
 
 func init() {
+	flag.Var(&KeyspacesToWatch, "keyspaces_to_watch", "Specifies which keyspaces this vtgate should have access to while routing queries or accessing the vschema")
 	flag.Var(&tabletFilters, "tablet_filters", "Specifies a comma-separated list of 'keyspace|shard_name or keyrange' values to filter the tablets to watch")
 	topoproto.TabletTypeListVar(&allowedTabletTypes, "allowed_tablet_types", "Specifies the tablet types this vtgate is allowed to route queries to")
-	RegisterGatewayCreator(gatewayImplementationDiscovery, createDiscoveryGateway)
 }
 
-type discoveryGateway struct {
+type tabletGateway struct {
 	queryservice.QueryService
 	hc            discovery.HealthCheck
 	tsc           *discovery.TabletStatsCache
@@ -87,11 +88,7 @@ type discoveryGateway struct {
 	buffer *buffer.Buffer
 }
 
-func createDiscoveryGateway(ctx context.Context, hc discovery.HealthCheck, serv srvtopo.Server, cell string, retryCount int) Gateway {
-	return NewDiscoveryGateway(ctx, hc, serv, cell, retryCount)
-}
-
-func NewDiscoveryGateway(ctx context.Context, hc discovery.HealthCheck, serv srvtopo.Server, cell string, retryCount int) *discoveryGateway {
+func NewTabletGateway(ctx context.Context, hc discovery.HealthCheck, serv srvtopo.Server, cell string, retryCount int) *tabletGateway {
 	var topoServer *topo.Server
 	if serv != nil {
 		var err error
@@ -101,7 +98,7 @@ func NewDiscoveryGateway(ctx context.Context, hc discovery.HealthCheck, serv srv
 		}
 	}
 
-	dg := &discoveryGateway{
+	dg := &tabletGateway{
 		hc:                hc,
 		tsc:               discovery.NewTabletStatsCacheDoNotSetListener(topoServer, cell),
 		srvTopoServer:     serv,
@@ -145,7 +142,7 @@ func NewDiscoveryGateway(ctx context.Context, hc discovery.HealthCheck, serv srv
 
 // RegisterStats registers the stats to export the lag since the last refresh
 // and the checksum of the topology
-func (dg *discoveryGateway) RegisterStats() {
+func (dg *tabletGateway) RegisterStats() {
 	stats.NewGaugeDurationFunc(
 		"TopologyWatcherMaxRefreshLag",
 		"maximum time since the topology watcher refreshed a cell",
@@ -161,7 +158,7 @@ func (dg *discoveryGateway) RegisterStats() {
 
 // topologyWatcherMaxRefreshLag returns the maximum lag since the watched
 // cells were refreshed from the topo server
-func (dg *discoveryGateway) topologyWatcherMaxRefreshLag() time.Duration {
+func (dg *tabletGateway) topologyWatcherMaxRefreshLag() time.Duration {
 	var lag time.Duration
 	for _, tw := range dg.tabletsWatchers {
 		cellLag := tw.RefreshLag()
@@ -173,7 +170,7 @@ func (dg *discoveryGateway) topologyWatcherMaxRefreshLag() time.Duration {
 }
 
 // topologyWatcherChecksum returns a checksum of the topology watcher state
-func (dg *discoveryGateway) topologyWatcherChecksum() int64 {
+func (dg *tabletGateway) topologyWatcherChecksum() int64 {
 	var checksum int64
 	for _, tw := range dg.tabletsWatchers {
 		checksum = checksum ^ int64(tw.TopoChecksum())
@@ -183,7 +180,7 @@ func (dg *discoveryGateway) topologyWatcherChecksum() int64 {
 
 // StatsUpdate forwards HealthCheck updates to TabletStatsCache and MasterBuffer.
 // It is part of the discovery.HealthCheckStatsListener interface.
-func (dg *discoveryGateway) StatsUpdate(ts *discovery.TabletStats) {
+func (dg *tabletGateway) StatsUpdate(ts *discovery.TabletStats) {
 	dg.tsc.StatsUpdate(ts)
 
 	if ts.Target.TabletType == topodatapb.TabletType_MASTER {
@@ -192,7 +189,7 @@ func (dg *discoveryGateway) StatsUpdate(ts *discovery.TabletStats) {
 }
 
 // WaitForTablets is part of the gateway.Gateway interface.
-func (dg *discoveryGateway) WaitForTablets(ctx context.Context, tabletTypesToWait []topodatapb.TabletType) error {
+func (dg *tabletGateway) WaitForTablets(ctx context.Context, tabletTypesToWait []topodatapb.TabletType) error {
 	// Skip waiting for tablets if we are not told to do so.
 	if len(tabletTypesToWait) == 0 {
 		return nil
@@ -209,7 +206,7 @@ func (dg *discoveryGateway) WaitForTablets(ctx context.Context, tabletTypesToWai
 
 // Close shuts down underlying connections.
 // This function hides the inner implementation.
-func (dg *discoveryGateway) Close(ctx context.Context) error {
+func (dg *tabletGateway) Close(ctx context.Context) error {
 	dg.buffer.Shutdown()
 	for _, ctw := range dg.tabletsWatchers {
 		ctw.Stop()
@@ -219,7 +216,7 @@ func (dg *discoveryGateway) Close(ctx context.Context) error {
 
 // CacheStatus returns a list of TabletCacheStatus per
 // keyspace/shard/tablet_type.
-func (dg *discoveryGateway) CacheStatus() TabletCacheStatusList {
+func (dg *tabletGateway) CacheStatus() TabletCacheStatusList {
 	dg.mu.RLock()
 	res := make(TabletCacheStatusList, 0, len(dg.statusAggregators))
 	for _, aggr := range dg.statusAggregators {
@@ -235,7 +232,7 @@ func (dg *discoveryGateway) CacheStatus() TabletCacheStatusList {
 // the middle of a transaction. While returning the error check if it maybe a result of
 // a resharding event, and set the re-resolve bit and let the upper layers
 // re-resolve and retry.
-func (dg *discoveryGateway) withRetry(ctx context.Context, target *querypb.Target, unused queryservice.QueryService, name string, inTransaction bool, inner func(ctx context.Context, target *querypb.Target, conn queryservice.QueryService) (bool, error)) error {
+func (dg *tabletGateway) withRetry(ctx context.Context, target *querypb.Target, unused queryservice.QueryService, name string, inTransaction bool, inner func(ctx context.Context, target *querypb.Target, conn queryservice.QueryService) (bool, error)) error {
 	var tabletLastUsed *topodatapb.Tablet
 	var err error
 	invalidTablets := make(map[string]bool)
@@ -374,13 +371,13 @@ func nextTablet(cell string, tablets []discovery.TabletStats, offset, length int
 	return -1
 }
 
-func (dg *discoveryGateway) updateStats(target *querypb.Target, startTime time.Time, err error) {
+func (dg *tabletGateway) updateStats(target *querypb.Target, startTime time.Time, err error) {
 	elapsed := time.Since(startTime)
 	aggr := dg.getStatsAggregator(target)
 	aggr.UpdateQueryInfo("", target.TabletType, elapsed, err != nil)
 }
 
-func (dg *discoveryGateway) getStatsAggregator(target *querypb.Target) *TabletStatusAggregator {
+func (dg *tabletGateway) getStatsAggregator(target *querypb.Target) *TabletStatusAggregator {
 	key := fmt.Sprintf("%v/%v/%v", target.Keyspace, target.Shard, target.TabletType.String())
 
 	// get existing aggregator
@@ -402,6 +399,32 @@ func (dg *discoveryGateway) getStatsAggregator(target *querypb.Target) *TabletSt
 	return aggr
 }
 
+// WaitForTablets is a helper method to wait for the provided tablets,
+// up until the *initialTabletTimeout. It will log what it is doing.
+// Note it has the same name as the tabletGateway's method, as it
+// just calls it.
+func WaitForTablets(gw *tabletGateway, tabletTypesToWait []topodatapb.TabletType) error {
+	log.Infof("Gateway waiting for serving tablets of types %v ...", tabletTypesToWait)
+	ctx, cancel := context.WithTimeout(context.Background(), *initialTabletTimeout)
+	defer cancel()
+
+	err := gw.WaitForTablets(ctx, tabletTypesToWait)
+	switch err {
+	case nil:
+		// Log so we know everything is fine.
+		log.Infof("Waiting for tablets completed")
+	case context.DeadlineExceeded:
+		// In this scenario, we were able to reach the
+		// topology service, but some tablets may not be
+		// ready. We just warn and keep going.
+		log.Warningf("Timeout waiting for all keyspaces / shards to have healthy tablets of types %v, may be in degraded mode", tabletTypesToWait)
+		err = nil
+	default:
+		// Nothing to do here, the caller will log.Fatalf.
+	}
+	return err
+}
+
 // NewShardError returns a new error with the shard info amended.
 func NewShardError(in error, target *querypb.Target, tablet *topodatapb.Tablet) error {
 	if in == nil {
diff --git a/go/vt/vtgate/tx_conn.go b/go/vt/vtgate/tx_conn.go
index ae8926f16b..1cf7b4c0c9 100644
--- a/go/vt/vtgate/tx_conn.go
+++ b/go/vt/vtgate/tx_conn.go
@@ -33,12 +33,12 @@ import (
 
 // TxConn is used for executing transactional requests.
 type TxConn struct {
-	gateway Gateway
+	gateway *tabletGateway
 	mode    vtgatepb.TransactionMode
 }
 
 // NewTxConn builds a new TxConn.
-func NewTxConn(gw Gateway, txMode vtgatepb.TransactionMode) *TxConn {
+func NewTxConn(gw *tabletGateway, txMode vtgatepb.TransactionMode) *TxConn {
 	return &TxConn{
 		gateway: gw,
 		mode:    txMode,
diff --git a/go/vt/vtgate/vcursor_impl.go b/go/vt/vtgate/vcursor_impl.go
index 3235086c4d..bba6596c4b 100644
--- a/go/vt/vtgate/vcursor_impl.go
+++ b/go/vt/vtgate/vcursor_impl.go
@@ -17,12 +17,9 @@ limitations under the License.
 package vtgate
 
 import (
-	"fmt"
 	"sync/atomic"
 	"time"
 
-	"vitess.io/vitess/go/vt/vtgate/planbuilder"
-
 	"golang.org/x/net/context"
 
 	"vitess.io/vitess/go/sqltypes"
@@ -37,11 +34,9 @@ import (
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	vtgatepb "vitess.io/vitess/go/vt/proto/vtgate"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
-	topoprotopb "vitess.io/vitess/go/vt/topo/topoproto"
 )
 
 var _ engine.VCursor = (*vcursorImpl)(nil)
-var _ planbuilder.ContextVSchema = (*vcursorImpl)(nil)
 var _ iExecute = (*Executor)(nil)
 
 // vcursor_impl needs these facilities to be able to be able to execute queries for vindexes
@@ -61,42 +56,33 @@ type vcursorImpl struct {
 	safeSession    *SafeSession
 	keyspace       string
 	tabletType     topodatapb.TabletType
-	destination    key.Destination
 	marginComments sqlparser.MarginComments
 	executor       iExecute
 	resolver       *srvtopo.Resolver
 	logStats       *LogStats
-	// rollbackOnPartialExec is set to true if any DML was successfully
+	// hasPartialDML is set to true if any DML was successfully
 	// executed. If there was a subsequent failure, the transaction
 	// must be forced to rollback.
-	rollbackOnPartialExec bool
-	vschema               *vindexes.VSchema
+	hasPartialDML bool
+	vschema       *vindexes.VSchema
 }
 
 // newVcursorImpl creates a vcursorImpl. Before creating this object, you have to separate out any marginComments that came with
 // the query and supply it here. Trailing comments are typically sent by the application for various reasons,
 // including as identifying markers. So, they have to be added back to all queries that are executed
 // on behalf of the original query.
-func newVCursorImpl(ctx context.Context, safeSession *SafeSession, marginComments sqlparser.MarginComments, executor *Executor, logStats *LogStats, vschema *vindexes.VSchema, resolver *srvtopo.Resolver) (*vcursorImpl, error) {
-	keyspace, tabletType, destination, err := parseDestinationTarget(safeSession.TargetString, vschema)
-	// ks
-	// select b from ts.a
-	// select b from ps.a
-	if err != nil {
-		return nil, err
-	}
+func newVCursorImpl(ctx context.Context, safeSession *SafeSession, keyspace string, tabletType topodatapb.TabletType, marginComments sqlparser.MarginComments, executor *Executor, logStats *LogStats, vschema *vindexes.VSchema, resolver *srvtopo.Resolver) *vcursorImpl {
 	return &vcursorImpl{
 		ctx:            ctx,
 		safeSession:    safeSession,
 		keyspace:       keyspace,
 		tabletType:     tabletType,
-		destination:    destination,
 		marginComments: marginComments,
 		executor:       executor,
 		logStats:       logStats,
 		vschema:        vschema,
 		resolver:       resolver,
-	}, nil
+	}
 }
 
 // Context returns the current Context.
@@ -174,7 +160,7 @@ func (vc *vcursorImpl) TargetString() string {
 }
 
 // Execute is part of the engine.VCursor interface.
-func (vc *vcursorImpl) Execute(method string, query string, bindVars map[string]*querypb.BindVariable, rollbackOnError bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error) {
+func (vc *vcursorImpl) Execute(method string, query string, bindVars map[string]*querypb.BindVariable, isDML bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error) {
 	session := vc.safeSession
 	if co == vtgatepb.CommitOrder_AUTOCOMMIT {
 		// For autocommit, we have to create an independent session.
@@ -185,19 +171,19 @@ func (vc *vcursorImpl) Execute(method string, query string, bindVars map[string]
 	}
 
 	qr, err := vc.executor.Execute(vc.ctx, method, session, vc.marginComments.Leading+query+vc.marginComments.Trailing, bindVars)
-	if err == nil && rollbackOnError {
-		vc.rollbackOnPartialExec = true
+	if err == nil && isDML {
+		vc.hasPartialDML = true
 	}
 	return qr, err
 }
 
 // ExecuteMultiShard is part of the engine.VCursor interface.
-func (vc *vcursorImpl) ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, rollbackOnError, autocommit bool) (*sqltypes.Result, []error) {
+func (vc *vcursorImpl) ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, isDML, autocommit bool) (*sqltypes.Result, []error) {
 	atomic.AddUint32(&vc.logStats.ShardQueries, uint32(len(queries)))
 	qr, errs := vc.executor.ExecuteMultiShard(vc.ctx, rss, commentedShardQueries(queries, vc.marginComments), vc.tabletType, vc.safeSession, false, autocommit)
 
-	if errs == nil && rollbackOnError {
-		vc.rollbackOnPartialExec = true
+	if errs == nil && isDML {
+		vc.hasPartialDML = true
 	}
 	return qr, errs
 }
@@ -229,7 +215,7 @@ func (vc *vcursorImpl) StreamExecuteMulti(query string, rss []*srvtopo.ResolvedS
 }
 
 // ExecuteKeyspaceID is part of the engine.VCursor interface.
-func (vc *vcursorImpl) ExecuteKeyspaceID(keyspace string, ksid []byte, query string, bindVars map[string]*querypb.BindVariable, rollbackOnError, autocommit bool) (*sqltypes.Result, error) {
+func (vc *vcursorImpl) ExecuteKeyspaceID(keyspace string, ksid []byte, query string, bindVars map[string]*querypb.BindVariable, isDML, autocommit bool) (*sqltypes.Result, error) {
 	atomic.AddUint32(&vc.logStats.ShardQueries, 1)
 	rss, _, err := vc.ResolveDestinations(keyspace, nil, []key.Destination{key.DestinationKeyspaceID(ksid)})
 	if err != nil {
@@ -239,11 +225,11 @@ func (vc *vcursorImpl) ExecuteKeyspaceID(keyspace string, ksid []byte, query str
 		Sql:           query,
 		BindVariables: bindVars,
 	}}
-	qr, errs := vc.ExecuteMultiShard(rss, queries, rollbackOnError, autocommit)
+	qr, errs := vc.ExecuteMultiShard(rss, queries, isDML, autocommit)
 
 	if len(errs) == 0 {
-		if rollbackOnError {
-			vc.rollbackOnPartialExec = true
+		if isDML {
+			vc.hasPartialDML = true
 		}
 		return qr, nil
 	}
@@ -254,16 +240,6 @@ func (vc *vcursorImpl) ResolveDestinations(keyspace string, ids []*querypb.Value
 	return vc.resolver.ResolveDestinations(vc.ctx, keyspace, vc.tabletType, ids, destinations)
 }
 
-// Destination implements the ContextVSchema interface
-func (vc *vcursorImpl) Destination() key.Destination {
-	return vc.destination
-}
-
-// TabletType implements the ContextVSchema interface
-func (vc *vcursorImpl) TabletType() topodatapb.TabletType {
-	return vc.tabletType
-}
-
 func commentedShardQueries(shardQueries []*querypb.BoundQuery, marginComments sqlparser.MarginComments) []*querypb.BoundQuery {
 	if marginComments.Leading == "" && marginComments.Trailing == "" {
 		return shardQueries
@@ -277,38 +253,3 @@ func commentedShardQueries(shardQueries []*querypb.BoundQuery, marginComments sq
 	}
 	return newQueries
 }
-
-// TargetDestination implements the ContextVSchema interface
-func (vc *vcursorImpl) TargetDestination(qualifier string) (key.Destination, *vindexes.Keyspace, topodatapb.TabletType, error) {
-	keyspaceName := vc.keyspace
-	if vc.destination == nil && qualifier != "" {
-		keyspaceName = qualifier
-	}
-	if keyspaceName == "" {
-		return nil, nil, 0, vterrors.New(vtrpcpb.Code_INVALID_ARGUMENT, "keyspace not specified")
-	}
-	keyspace := vc.vschema.Keyspaces[keyspaceName]
-	if keyspace == nil {
-		return nil, nil, 0, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "no keyspace with name [%s] found", keyspaceName)
-	}
-	return vc.destination, keyspace.Keyspace, vc.tabletType, nil
-}
-
-// ParseDestinationTarget parses destination target string and sets default keyspace if possible.
-func parseDestinationTarget(targetString string, vschema *vindexes.VSchema) (string, topodatapb.TabletType, key.Destination, error) {
-	destKeyspace, destTabletType, dest, err := topoprotopb.ParseDestination(targetString, defaultTabletType)
-	// Set default keyspace
-	if destKeyspace == "" && len(vschema.Keyspaces) == 1 {
-		for k := range vschema.Keyspaces {
-			destKeyspace = k
-		}
-	}
-	return destKeyspace, destTabletType, dest, err
-}
-
-func (vc *vcursorImpl) planPrefixKey() string {
-	if vc.destination != nil {
-		return fmt.Sprintf("%s%s%s", vc.keyspace, vindexes.TabletTypeSuffix[vc.tabletType], vc.destination.String())
-	}
-	return fmt.Sprintf("%s%s", vc.keyspace, vindexes.TabletTypeSuffix[vc.tabletType])
-}
diff --git a/go/vt/vtgate/vindexes/consistent_lookup.go b/go/vt/vtgate/vindexes/consistent_lookup.go
index 84072e75be..1c64d218cc 100644
--- a/go/vt/vtgate/vindexes/consistent_lookup.go
+++ b/go/vt/vtgate/vindexes/consistent_lookup.go
@@ -266,19 +266,19 @@ func (lu *clCommon) handleDup(vcursor VCursor, values []sqltypes.Value, ksid []b
 	bindVars[lu.lkp.To] = sqltypes.BytesBindVariable(ksid)
 
 	// Lock the lookup row using pre priority.
-	qr, err := vcursor.Execute("VindexCreate", lu.lockLookupQuery, bindVars, false /* rollbackOnError */, vtgatepb.CommitOrder_PRE)
+	qr, err := vcursor.Execute("VindexCreate", lu.lockLookupQuery, bindVars, false /* isDML */, vtgatepb.CommitOrder_PRE)
 	if err != nil {
 		return err
 	}
 	switch len(qr.Rows) {
 	case 0:
-		if _, err := vcursor.Execute("VindexCreate", lu.insertLookupQuery, bindVars, true /* rollbackOnError */, vtgatepb.CommitOrder_PRE); err != nil {
+		if _, err := vcursor.Execute("VindexCreate", lu.insertLookupQuery, bindVars, true /* isDML */, vtgatepb.CommitOrder_PRE); err != nil {
 			return err
 		}
 	case 1:
 		existingksid := qr.Rows[0][0].ToBytes()
 		// Lock the target row using normal transaction priority.
-		qr, err = vcursor.ExecuteKeyspaceID(lu.keyspace, existingksid, lu.lockOwnerQuery, bindVars, false /* rollbackOnError */, false /* autocommit */)
+		qr, err = vcursor.ExecuteKeyspaceID(lu.keyspace, existingksid, lu.lockOwnerQuery, bindVars, false /* isDML */, false /* autocommit */)
 		if err != nil {
 			return err
 		}
@@ -288,7 +288,7 @@ func (lu *clCommon) handleDup(vcursor VCursor, values []sqltypes.Value, ksid []b
 		if bytes.Equal(existingksid, ksid) {
 			return nil
 		}
-		if _, err := vcursor.Execute("VindexCreate", lu.updateLookupQuery, bindVars, true /* rollbackOnError */, vtgatepb.CommitOrder_PRE); err != nil {
+		if _, err := vcursor.Execute("VindexCreate", lu.updateLookupQuery, bindVars, true /* isDML */, vtgatepb.CommitOrder_PRE); err != nil {
 			return err
 		}
 	default:
diff --git a/go/vt/vtgate/vindexes/lookup_internal.go b/go/vt/vtgate/vindexes/lookup_internal.go
index 2ea8a77e45..bfcc38a2c1 100644
--- a/go/vt/vtgate/vindexes/lookup_internal.go
+++ b/go/vt/vtgate/vindexes/lookup_internal.go
@@ -76,7 +76,7 @@ func (lkp *lookupInternal) Lookup(vcursor VCursor, ids []sqltypes.Value) ([]*sql
 		if lkp.Autocommit {
 			co = vtgatepb.CommitOrder_AUTOCOMMIT
 		}
-		result, err = vcursor.Execute("VindexLookup", lkp.sel, bindVars, false /* rollbackOnError */, co)
+		result, err = vcursor.Execute("VindexLookup", lkp.sel, bindVars, false /* isDML */, co)
 		if err != nil {
 			return nil, fmt.Errorf("lookup.Map: %v", err)
 		}
@@ -101,7 +101,7 @@ func (lkp *lookupInternal) VerifyCustom(vcursor VCursor, ids, values []sqltypes.
 			lkp.FromColumns[0]: sqltypes.ValueBindVariable(id),
 			lkp.To:             sqltypes.ValueBindVariable(values[i]),
 		}
-		result, err := vcursor.Execute("VindexVerify", lkp.ver, bindVars, false /* rollbackOnError */, co)
+		result, err := vcursor.Execute("VindexVerify", lkp.ver, bindVars, false /* isDML */, co)
 		if err != nil {
 			return nil, fmt.Errorf("lookup.Verify: %v", err)
 		}
@@ -206,7 +206,7 @@ func (lkp *lookupInternal) createCustom(vcursor VCursor, rowsColValues [][]sqlty
 		fmt.Fprintf(buf, "%s=values(%s)", lkp.To, lkp.To)
 	}
 
-	if _, err := vcursor.Execute("VindexCreate", buf.String(), bindVars, true /* rollbackOnError */, co); err != nil {
+	if _, err := vcursor.Execute("VindexCreate", buf.String(), bindVars, true /* isDML */, co); err != nil {
 		return fmt.Errorf("lookup.Create: %v", err)
 	}
 	return nil
@@ -247,7 +247,7 @@ func (lkp *lookupInternal) Delete(vcursor VCursor, rowsColValues [][]sqltypes.Va
 			bindVars[lkp.FromColumns[colIdx]] = sqltypes.ValueBindVariable(columnValue)
 		}
 		bindVars[lkp.To] = sqltypes.ValueBindVariable(value)
-		_, err := vcursor.Execute("VindexDelete", lkp.del, bindVars, true /* rollbackOnError */, co)
+		_, err := vcursor.Execute("VindexDelete", lkp.del, bindVars, true /* isDML */, co)
 		if err != nil {
 			return fmt.Errorf("lookup.Delete: %v", err)
 		}
diff --git a/go/vt/vtgate/vindexes/vindex.go b/go/vt/vtgate/vindexes/vindex.go
index 0fea9233bb..ce7951aec6 100644
--- a/go/vt/vtgate/vindexes/vindex.go
+++ b/go/vt/vtgate/vindexes/vindex.go
@@ -35,8 +35,8 @@ import (
 // in the current context and session of a VTGate request. Vindexes
 // can use this interface to execute lookup queries.
 type VCursor interface {
-	Execute(method string, query string, bindvars map[string]*querypb.BindVariable, rollbackOnError bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error)
-	ExecuteKeyspaceID(keyspace string, ksid []byte, query string, bindVars map[string]*querypb.BindVariable, rollbackOnError, autocommit bool) (*sqltypes.Result, error)
+	Execute(method string, query string, bindvars map[string]*querypb.BindVariable, isDML bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error)
+	ExecuteKeyspaceID(keyspace string, ksid []byte, query string, bindVars map[string]*querypb.BindVariable, isDML, autocommit bool) (*sqltypes.Result, error)
 }
 
 // Vindex defines the interface required to register a vindex.
diff --git a/go/vt/vtgate/vtgate.go b/go/vt/vtgate/vtgate.go
index 0a92c6dbf9..662dd64d6a 100644
--- a/go/vt/vtgate/vtgate.go
+++ b/go/vt/vtgate/vtgate.go
@@ -101,7 +101,7 @@ type VTGate struct {
 	resolver *Resolver
 	vsm      *vstreamManager
 	txConn   *TxConn
-	gw       Gateway
+	gw       *tabletGateway
 
 	// stats objects.
 	// TODO(sougou): This needs to be cleaned up. There
@@ -133,7 +133,7 @@ func Init(ctx context.Context, hc discovery.HealthCheck, serv srvtopo.Server, ce
 	// Build objects from low to high level.
 	// Start with the gateway. If we can't reach the topology service,
 	// we can't go on much further, so we log.Fatal out.
-	gw := GatewayCreator()(ctx, hc, serv, cell, retryCount)
+	gw := NewTabletGateway(ctx, hc, serv, cell, retryCount)
 	gw.RegisterStats()
 	if err := WaitForTablets(gw, tabletTypesToWait); err != nil {
 		log.Fatalf("gateway.WaitForTablets failed: %v", err)
@@ -227,7 +227,7 @@ func (vtg *VTGate) IsHealthy() error {
 }
 
 // Gateway returns the current gateway implementation. Mostly used for tests.
-func (vtg *VTGate) Gateway() Gateway {
+func (vtg *VTGate) Gateway() *tabletGateway {
 	return vtg.gw
 }
 
diff --git a/go/vt/vttablet/queryservice/wrapped.go b/go/vt/vttablet/queryservice/wrapped.go
index 185046dd9d..9443951d6f 100644
--- a/go/vt/vttablet/queryservice/wrapped.go
+++ b/go/vt/vttablet/queryservice/wrapped.go
@@ -39,7 +39,7 @@ type WrapperFunc func(ctx context.Context, target *querypb.Target, conn QuerySer
 // Wrap returns a wrapped version of the original QueryService implementation.
 // This lets you avoid repeating boiler-plate code by consolidating it in the
 // wrapper function.
-// A good example of this is go/vt/vtgate/gateway/discoverygateway.go.
+// A good example of this is go/vt/vtgate/gateway/tabletgateway.go.
 // For every method invocation, the wrapper function is called, which can
 // in turn call the provided inner function that will use the input parameters
 // to call the implementation. In order to load balance across multiple
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vplayer.go b/go/vt/vttablet/tabletmanager/vreplication/vplayer.go
index 7c6794ad45..7b14d33a3b 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vplayer.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vplayer.go
@@ -21,7 +21,6 @@ import (
 	"fmt"
 	"io"
 	"math"
-	"strings"
 	"time"
 
 	"golang.org/x/net/context"
@@ -291,6 +290,7 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
 	// TODO(sougou): if we also stored the time of the last event, we
 	// can estimate this value more accurately.
 	defer vp.vr.stats.SecondsBehindMaster.Set(math.MaxInt64)
+	var sbm int64 = -1
 	for {
 		items, err := relay.Fetch()
 		if err != nil {
@@ -323,7 +323,7 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
 				if event.Timestamp != 0 {
 					vp.lastTimestampNs = event.Timestamp * 1e9
 					vp.timeOffsetNs = time.Now().UnixNano() - event.CurrentTime
-					vp.vr.stats.SecondsBehindMaster.Set(event.CurrentTime/1e9 - event.Timestamp)
+					sbm = event.CurrentTime/1e9 - event.Timestamp
 				}
 				mustSave := false
 				switch event.Type {
@@ -349,6 +349,10 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
 				}
 			}
 		}
+		if sbm >= 0 {
+			vp.vr.stats.SecondsBehindMaster.Set(sbm)
+		}
+
 	}
 }
 
@@ -420,18 +424,15 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 		stats.Send(fmt.Sprintf("%v", event.FieldEvent))
 
 	case binlogdatapb.VEventType_INSERT, binlogdatapb.VEventType_DELETE, binlogdatapb.VEventType_UPDATE, binlogdatapb.VEventType_REPLACE:
-		// If the event is for one of the AWS RDS "special" tables, we skip
-		if !strings.Contains(event.Dml, " mysql.rds_") {
-			// This is a player using stament based replication
-			if err := vp.vr.dbClient.Begin(); err != nil {
-				return err
-			}
+		// This is a player using stament based replication
+		if err := vp.vr.dbClient.Begin(); err != nil {
+			return err
+		}
 
-			if err := vp.applyStmtEvent(ctx, event); err != nil {
-				return err
-			}
-			stats.Send(fmt.Sprintf(event.Dml))
+		if err := vp.applyStmtEvent(ctx, event); err != nil {
+			return err
 		}
+		stats.Send(fmt.Sprintf(event.Dml))
 	case binlogdatapb.VEventType_ROW:
 		// This player is configured for row based replication
 		if err := vp.vr.dbClient.Begin(); err != nil {
diff --git a/go/vt/vttablet/tabletserver/messager/message_manager.go b/go/vt/vttablet/tabletserver/messager/message_manager.go
index c3bb38f65b..f418d8557e 100644
--- a/go/vt/vttablet/tabletserver/messager/message_manager.go
+++ b/go/vt/vttablet/tabletserver/messager/message_manager.go
@@ -165,8 +165,6 @@ type messageManager struct {
 	fieldResult  *sqltypes.Result
 	ackWaitTime  time.Duration
 	purgeAfter   time.Duration
-	minBackoff   time.Duration
-	maxBackoff   time.Duration
 	batchSize    int
 	pollerTicks  *timer.Timer
 	purgeTicks   *timer.Timer
@@ -225,8 +223,6 @@ func newMessageManager(tsv TabletService, vs VStreamer, table *schema.Table, pos
 		},
 		ackWaitTime:     table.MessageInfo.AckWaitDuration,
 		purgeAfter:      table.MessageInfo.PurgeAfterDuration,
-		minBackoff:      table.MessageInfo.MinBackoff,
-		maxBackoff:      table.MessageInfo.MaxBackoff,
 		batchSize:       table.MessageInfo.BatchSize,
 		cache:           newCache(table.MessageInfo.CacheSize),
 		pollerTicks:     timer.NewTimer(table.MessageInfo.PollInterval),
@@ -250,19 +246,11 @@ func newMessageManager(tsv TabletService, vs VStreamer, table *schema.Table, pos
 	mm.ackQuery = sqlparser.BuildParsedQuery(
 		"update %v set time_acked = %a, time_next = null where id in %a and time_acked is null",
 		mm.name, ":time_acked", "::ids")
+	mm.postponeQuery = sqlparser.BuildParsedQuery(
+		"update %v set time_next = %a+(%a<<ifnull(epoch, 0)), epoch = ifnull(epoch, 0)+1 where id in %a and time_acked is null",
+		mm.name, ":time_now", ":wait_time", "::ids")
 	mm.purgeQuery = sqlparser.BuildParsedQuery(
 		"delete from %v where time_acked < %a limit 500", mm.name, ":time_acked")
-
-	// if a maxBackoff is set, incorporate it into the update statement
-	if mm.maxBackoff > 0 {
-		mm.postponeQuery = sqlparser.BuildParsedQuery(
-			"update %v set time_next = %a+if(%a<<ifnull(epoch, 0) > %a, %a, %a<<ifnull(epoch, 0)), epoch = ifnull(epoch, 0)+1 where id in %a and time_acked is null",
-			mm.name, ":time_now", ":min_backoff", ":max_backoff", ":max_backoff", ":min_backoff", "::ids")
-	} else {
-		mm.postponeQuery = sqlparser.BuildParsedQuery(
-			"update %v set time_next = %a+(%a<<ifnull(epoch, 0)), epoch = ifnull(epoch, 0)+1 where id in %a and time_acked is null",
-			mm.name, ":time_now", ":min_backoff", "::ids")
-	}
 	return mm
 }
 
@@ -804,18 +792,11 @@ func (mm *messageManager) GeneratePostponeQuery(ids []string) (string, map[strin
 			Value: []byte(id),
 		})
 	}
-
-	bvs := map[string]*querypb.BindVariable{
-		"time_now":    sqltypes.Int64BindVariable(time.Now().UnixNano()),
-		"min_backoff": sqltypes.Int64BindVariable(int64(mm.minBackoff)),
-		"ids":         idbvs,
-	}
-
-	if mm.maxBackoff > 0 {
-		bvs["max_backoff"] = sqltypes.Int64BindVariable(int64(mm.maxBackoff))
+	return mm.postponeQuery.Query, map[string]*querypb.BindVariable{
+		"time_now":  sqltypes.Int64BindVariable(time.Now().UnixNano()),
+		"wait_time": sqltypes.Int64BindVariable(int64(mm.ackWaitTime)),
+		"ids":       idbvs,
 	}
-
-	return mm.postponeQuery.Query, bvs
 }
 
 // GeneratePurgeQuery returns the query and bind vars for purging messages.
diff --git a/go/vt/vttablet/tabletserver/schema/load_table.go b/go/vt/vttablet/tabletserver/schema/load_table.go
index ce3b2f3f88..e46868e115 100644
--- a/go/vt/vttablet/tabletserver/schema/load_table.go
+++ b/go/vt/vttablet/tabletserver/schema/load_table.go
@@ -100,16 +100,6 @@ func loadMessageInfo(ta *Table, comment string) error {
 	if ta.MessageInfo.PollInterval, err = getDuration(keyvals, "vt_poller_interval"); err != nil {
 		return err
 	}
-
-	// errors are ignored because these fields are optional and 0 is the default value
-	ta.MessageInfo.MinBackoff, _ = getDuration(keyvals, "vt_min_backoff")
-	// the original default minimum backoff was based on ack wait timeout, so this preserves that
-	if ta.MessageInfo.MinBackoff == 0 {
-		ta.MessageInfo.MinBackoff = ta.MessageInfo.AckWaitDuration
-	}
-
-	ta.MessageInfo.MaxBackoff, _ = getDuration(keyvals, "vt_max_backoff")
-
 	for _, col := range requiredCols {
 		num := ta.FindColumn(sqlparser.NewColIdent(col))
 		if num == -1 {
diff --git a/go/vt/vttablet/tabletserver/schema/schema.go b/go/vt/vttablet/tabletserver/schema/schema.go
index ebf622bddb..c99a279dd3 100644
--- a/go/vt/vttablet/tabletserver/schema/schema.go
+++ b/go/vt/vttablet/tabletserver/schema/schema.go
@@ -92,14 +92,6 @@ type MessageInfo struct {
 	// PollInterval specifies the polling frequency to
 	// look for messages to be sent.
 	PollInterval time.Duration
-
-	// MinBackoff specifies the shortest duration message manager
-	// should wait before rescheduling a message
-	MinBackoff time.Duration
-
-	// MaxBackoff specifies the longest duration message manager
-	// should wait before rescheduling a message
-	MaxBackoff time.Duration
 }
 
 // NewTable creates a new Table.
diff --git a/go/vt/wrangler/stream_migrater.go b/go/vt/wrangler/stream_migrater.go
index e5edad9087..963f6e711c 100644
--- a/go/vt/wrangler/stream_migrater.go
+++ b/go/vt/wrangler/stream_migrater.go
@@ -18,12 +18,13 @@ package wrangler
 
 import (
 	"fmt"
-	"github.com/golang/protobuf/proto"
-	"golang.org/x/net/context"
 	"sort"
 	"strings"
 	"sync"
 	"text/template"
+
+	"github.com/golang/protobuf/proto"
+	"golang.org/x/net/context"
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/binlog/binlogplayer"
diff --git a/tools/dependency_check.sh b/tools/dependency_check.sh
index 058e702ae6..7d5179c161 100755
--- a/tools/dependency_check.sh
+++ b/tools/dependency_check.sh
@@ -24,6 +24,6 @@ function fail() {
 # These binaries are required to 'make test'
 # mysqld might be in /usr/sbin which will not be in the default PATH
 PATH="/usr/sbin:$PATH"
-for binary in k3s mysqld consul etcd etcdctl zksrv.sh javadoc mvn ant curl wget zip unzip; do
+for binary in mysqld consul etcd etcdctl zksrv.sh javadoc mvn ant curl wget zip unzip; do
   command -v "$binary" > /dev/null || fail "${binary} is not installed in PATH. See https://vitess.io/contributing/build-from-source for install instructions."
 done;


Test patch:
diff --git a/examples/compose/vtcompose/docker-compose.test.yml b/examples/compose/vtcompose/docker-compose.test.yml
deleted file mode 100644
index 1e55db8e9f..0000000000
--- a/examples/compose/vtcompose/docker-compose.test.yml
+++ /dev/null
@@ -1,318 +0,0 @@
-# This file was generated by running `go run vtcompose/vtcompose.go` with no arguments.
-
-services:
-  consul1:
-    command: agent -server -bootstrap-expect 3 -ui -disable-host-node-id -client 0.0.0.0
-    hostname: consul1
-    image: consul:latest
-    ports:
-      - 8400:8400
-      - 8500:8500
-      - 8600:8600
-  consul2:
-    command: agent -server -retry-join consul1 -disable-host-node-id
-    depends_on:
-      - consul1
-    expose:
-      - "8400"
-      - "8500"
-      - "8600"
-    hostname: consul2
-    image: consul:latest
-  consul3:
-    command: agent -server -retry-join consul1 -disable-host-node-id
-    depends_on:
-      - consul1
-    expose:
-      - "8400"
-      - "8500"
-      - "8600"
-    hostname: consul3
-    image: consul:latest
-  schemaload_test_keyspace:
-    command:
-      - sh
-      - -c
-      - /script/schemaload.sh
-    depends_on:
-      vttablet101:
-        condition: service_healthy
-      vttablet201:
-        condition: service_healthy
-    environment:
-      - TOPOLOGY_FLAGS=-topo_implementation consul -topo_global_server_address consul1:8500
-        -topo_global_root vitess/global
-      - WEB_PORT=8080
-      - GRPC_PORT=15999
-      - CELL=test
-      - KEYSPACE=test_keyspace
-      - TARGETTAB=test-0000000101
-      - SLEEPTIME=15
-      - VSCHEMA_FILE=test_keyspace_vschema.json
-      - SCHEMA_FILES=test_keyspace_schema_file.sql
-      - POST_LOAD_FILE=
-      - EXTERNAL_DB=0
-    image: vitess/base
-    volumes:
-      - .:/script
-  schemaload_unsharded_keyspace:
-    command:
-      - sh
-      - -c
-      - /script/schemaload.sh
-    depends_on:
-      vttablet301:
-        condition: service_healthy
-    environment:
-      - TOPOLOGY_FLAGS=-topo_implementation consul -topo_global_server_address consul1:8500
-        -topo_global_root vitess/global
-      - WEB_PORT=8080
-      - GRPC_PORT=15999
-      - CELL=test
-      - KEYSPACE=unsharded_keyspace
-      - TARGETTAB=test-0000000301
-      - SLEEPTIME=15
-      - VSCHEMA_FILE=unsharded_keyspace_vschema.json
-      - SCHEMA_FILES=unsharded_keyspace_schema_file.sql
-      - POST_LOAD_FILE=
-      - EXTERNAL_DB=0
-    image: vitess/base
-    volumes:
-      - .:/script
-  vtctld:
-    command:
-      - sh
-      - -c
-      - ' $$VTROOT/bin/vtctld -topo_implementation consul -topo_global_server_address
-      consul1:8500 -topo_global_root vitess/global -cell test -workflow_manager_init
-      -workflow_manager_use_election -service_map ''grpc-vtctl'' -backup_storage_implementation
-      file -file_backup_storage_root $$VTDATAROOT/backups -logtostderr=true -port
-      8080 -grpc_port 15999 -pid_file $$VTDATAROOT/tmp/vtctld.pid '
-    depends_on:
-      - consul1
-      - consul2
-      - consul3
-    image: vitess/base
-    ports:
-      - 15000:8080
-      - "15999"
-    volumes:
-      - .:/script
-  vtgate:
-    command:
-      - sh
-      - -c
-      - '/script/run-forever.sh $$VTROOT/bin/vtgate -topo_implementation consul -topo_global_server_address
-      consul1:8500 -topo_global_root vitess/global -logtostderr=true -port 8080 -grpc_port
-      15999 -mysql_server_port 15306 -mysql_auth_server_impl none -cell test -cells_to_watch
-      test -tablet_types_to_wait MASTER,REPLICA,RDONLY -gateway_implementation discoverygateway
-      -service_map ''grpc-vtgateservice'' -pid_file $$VTDATAROOT/tmp/vtgate.pid -normalize_queries=true '
-    depends_on:
-      - vtctld
-    image: vitess/base
-    ports:
-      - 15099:8080
-      - "15999"
-      - 15306:15306
-    volumes:
-      - .:/script
-  vttablet101:
-    command:
-      - sh
-      - -c
-      - /script/vttablet-up.sh 101
-    depends_on:
-      - vtctld
-    environment:
-      - TOPOLOGY_FLAGS=-topo_implementation consul -topo_global_server_address consul1:8500
-        -topo_global_root vitess/global
-      - WEB_PORT=8080
-      - GRPC_PORT=15999
-      - CELL=test
-      - KEYSPACE=test_keyspace
-      - SHARD=-80
-      - ROLE=master
-      - VTHOST=vttablet101
-      - EXTERNAL_DB=0
-      - DB_PORT=
-      - DB_HOST=
-      - DB_USER=
-      - DB_PASS=
-      - DB_CHARSET=
-    healthcheck:
-      interval: 30s
-      retries: 15
-      test:
-        - CMD-SHELL
-        - curl localhost:8080/debug/health
-      timeout: 10s
-    image: vitess/base
-    ports:
-      - 15101:8080
-      - "15999"
-      - "3306"
-    volumes:
-      - .:/script
-  vttablet102:
-    command:
-      - sh
-      - -c
-      - /script/vttablet-up.sh 102
-    depends_on:
-      - vtctld
-    environment:
-      - TOPOLOGY_FLAGS=-topo_implementation consul -topo_global_server_address consul1:8500
-        -topo_global_root vitess/global
-      - WEB_PORT=8080
-      - GRPC_PORT=15999
-      - CELL=test
-      - KEYSPACE=test_keyspace
-      - SHARD=-80
-      - ROLE=replica
-      - VTHOST=vttablet102
-      - EXTERNAL_DB=0
-      - DB_PORT=
-      - DB_HOST=
-      - DB_USER=
-      - DB_PASS=
-      - DB_CHARSET=
-    healthcheck:
-      interval: 30s
-      retries: 15
-      test:
-        - CMD-SHELL
-        - curl localhost:8080/debug/health
-      timeout: 10s
-    image: vitess/base
-    ports:
-      - 15102:8080
-      - "15999"
-      - "3306"
-    volumes:
-      - .:/script
-  vttablet201:
-    command:
-      - sh
-      - -c
-      - /script/vttablet-up.sh 201
-    depends_on:
-      - vtctld
-    environment:
-      - TOPOLOGY_FLAGS=-topo_implementation consul -topo_global_server_address consul1:8500
-        -topo_global_root vitess/global
-      - WEB_PORT=8080
-      - GRPC_PORT=15999
-      - CELL=test
-      - KEYSPACE=test_keyspace
-      - SHARD=80-
-      - ROLE=master
-      - VTHOST=vttablet201
-      - EXTERNAL_DB=0
-      - DB_PORT=
-      - DB_HOST=
-      - DB_USER=
-      - DB_PASS=
-      - DB_CHARSET=
-    healthcheck:
-      interval: 30s
-      retries: 15
-      test:
-        - CMD-SHELL
-        - curl localhost:8080/debug/health
-      timeout: 10s
-    image: vitess/base
-    ports:
-      - 15201:8080
-      - "15999"
-      - "3306"
-    volumes:
-      - .:/script
-  vttablet202:
-    command:
-      - sh
-      - -c
-      - /script/vttablet-up.sh 202
-    depends_on:
-      - vtctld
-    environment:
-      - TOPOLOGY_FLAGS=-topo_implementation consul -topo_global_server_address consul1:8500
-        -topo_global_root vitess/global
-      - WEB_PORT=8080
-      - GRPC_PORT=15999
-      - CELL=test
-      - KEYSPACE=test_keyspace
-      - SHARD=80-
-      - ROLE=replica
-      - VTHOST=vttablet202
-      - EXTERNAL_DB=0
-      - DB_PORT=
-      - DB_HOST=
-      - DB_USER=
-      - DB_PASS=
-      - DB_CHARSET=
-    healthcheck:
-      interval: 30s
-      retries: 15
-      test:
-        - CMD-SHELL
-        - curl localhost:8080/debug/health
-      timeout: 10s
-    image: vitess/base
-    ports:
-      - 15202:8080
-      - "15999"
-      - "3306"
-    volumes:
-      - .:/script
-  vttablet301:
-    command:
-      - sh
-      - -c
-      - /script/vttablet-up.sh 301
-    depends_on:
-      - vtctld
-    environment:
-      - TOPOLOGY_FLAGS=-topo_implementation consul -topo_global_server_address consul1:8500
-        -topo_global_root vitess/global
-      - WEB_PORT=8080
-      - GRPC_PORT=15999
-      - CELL=test
-      - KEYSPACE=unsharded_keyspace
-      - SHARD=-
-      - ROLE=master
-      - VTHOST=vttablet301
-      - EXTERNAL_DB=0
-      - DB_PORT=
-      - DB_HOST=
-      - DB_USER=
-      - DB_PASS=
-      - DB_CHARSET=
-    healthcheck:
-      interval: 30s
-      retries: 15
-      test:
-        - CMD-SHELL
-        - curl localhost:8080/debug/health
-      timeout: 10s
-    image: vitess/base
-    ports:
-      - 15301:8080
-      - "15999"
-      - "3306"
-    volumes:
-      - .:/script
-  vtwork:
-    command:
-      - sh
-      - -c
-      - '$$VTROOT/bin/vtworker -topo_implementation consul -topo_global_server_address
-      consul1:8500 -topo_global_root vitess/global -cell test -logtostderr=true -service_map
-      ''grpc-vtworker'' -port 8080 -grpc_port 15999 -use_v3_resharding_mode=true -pid_file
-      $$VTDATAROOT/tmp/vtwork.pid '
-    depends_on:
-      - vtctld
-    image: vitess/base
-    ports:
-      - 15100:8080
-      - "15999"
-version: "2.1"
\ No newline at end of file
diff --git a/examples/compose/vtcompose/vtcompose_test.go b/examples/compose/vtcompose/vtcompose_test.go
deleted file mode 100644
index fefd595bcc..0000000000
--- a/examples/compose/vtcompose/vtcompose_test.go
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Copyright 2019 The Vitess Authors.
-
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
-
- *     http://www.apache.org/licenses/LICENSE-2.0
-
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package main
-
-import (
-	"fmt"
-	"github.com/stretchr/testify/assert"
-	"strconv"
-	"testing"
-)
-
-var (
-	referenceYaml         = string(readFile("./docker-compose.test.yml"))
-	testComposeFile       = readFile("./docker-compose.base.yml")
-	testKeyspaceInfoMap   = parseKeyspaceInfo(DefaultKeyspaceData)
-	testExternalDbInfoMap = parseExternalDbData(DefaultExternalDbData)
-
-	testVtOpts = vtOptions{
-		webPort:       DefaultWebPort,
-		gRpcPort:      DefaultGrpcPort,
-		mySqlPort:     DefaultMysqlPort,
-		topologyFlags: DefaultTopologyFlags,
-		cell:          DefaultCell,
-	}
-)
-
-func TestGenerateCorrectFileWithDefaultOpts(t *testing.T) {
-	baseFile := testComposeFile
-	finalFile := applyDockerComposePatches(baseFile, testKeyspaceInfoMap, testExternalDbInfoMap, testVtOpts)
-
-	yamlString := string(finalFile)
-	assert.YAMLEq(t, referenceYaml, yamlString)
-}
-
-func TestOptsAppliedThroughoutGeneratedFile(t *testing.T) {
-	baseFile := testComposeFile
-	options := vtOptions{
-		webPort:       55_555,
-		gRpcPort:      66_666,
-		mySqlPort:     77_777,
-		topologyFlags: "-custom -flags",
-		cell:          "custom cell",
-	}
-	finalFile := applyDockerComposePatches(baseFile, testKeyspaceInfoMap, testExternalDbInfoMap, options)
-	yamlString := string(finalFile)
-
-	// These asserts are not exhaustive, but should cover most cases.
-	assert.NotContains(t, yamlString, strconv.Itoa(DefaultWebPort))
-	assert.Contains(t, yamlString, strconv.Itoa(options.webPort))
-
-	assert.NotContains(t, yamlString, strconv.Itoa(DefaultGrpcPort))
-	assert.Contains(t, yamlString, strconv.Itoa(options.gRpcPort))
-
-	assert.NotContains(t, yamlString, ":"+strconv.Itoa(DefaultMysqlPort))
-	assert.Contains(t, yamlString, ":"+strconv.Itoa(options.webPort))
-
-	assert.NotContains(t, yamlString, fmt.Sprintf("-cell %s", DefaultCell))
-	assert.Contains(t, yamlString, fmt.Sprintf("-cell %s", options.cell))
-
-	assert.Contains(t, yamlString, fmt.Sprintf("- TOPOLOGY_FLAGS=%s", options.topologyFlags))
-	assert.NotContains(t, yamlString, DefaultTopologyFlags)
-}
diff --git a/go/test/utils/diff.go b/go/test/utils/diff.go
deleted file mode 100644
index 05a1f9eb66..0000000000
--- a/go/test/utils/diff.go
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
-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 utils
-
-import (
-	"testing"
-
-	"github.com/google/go-cmp/cmp"
-)
-
-// MustMatchFn is used to create a common diff function for a test file
-// Usage in *_test.go file:
-//
-// Top declaration:
-//
-// var mustMatch = testutils.MustMatchFn(
-// 	[]interface{}{  // types with unexported fields
-// 		type1{},
-// 		type2{},
-// 		...
-// 		typeN{},
-// 	},
-// 	[]string{  // ignored fields
-// 		".id",        // id numbers are unstable
-// 		".createAt",  // created dates might not be interesting to compare
-// 	},
-// )
-//
-// In Test*() function:
-//
-// mustMatch(t, want, got, "something doesn't match")
-func MustMatchFn(allowUnexportedTypes []interface{}, ignoredFields []string, extraOpts ...cmp.Option) func(t *testing.T, want, got interface{}, errMsg string) {
-	diffOpts := append([]cmp.Option{
-		cmp.AllowUnexported(allowUnexportedTypes...),
-		cmpIgnoreFields(ignoredFields...),
-	}, extraOpts...)
-	// Diffs want/got and fails with errMsg on any failure.
-	return func(t *testing.T, want, got interface{}, errMsg string) {
-		t.Helper()
-		diff := cmp.Diff(want, got, diffOpts...)
-		if diff != "" {
-			t.Fatalf("%s: (-want +got)\n%v", errMsg, diff)
-		}
-	}
-}
-
-// MustMatch is a convenience version of MustMatchFn with no overrides.
-// Usage in Test*() function:
-//
-// testutils.MustMatch(t, want, got, "something doesn't match")
-var MustMatch = MustMatchFn(nil, nil)
-
-// Skips fields of pathNames for cmp.Diff.
-// Similar to standard cmpopts.IgnoreFields, but allows unexported fields.
-func cmpIgnoreFields(pathNames ...string) cmp.Option {
-	skipFields := make(map[string]bool, len(pathNames))
-	for _, name := range pathNames {
-		skipFields[name] = true
-	}
-
-	return cmp.FilterPath(func(path cmp.Path) bool {
-		for _, ps := range path {
-			if skipFields[ps.String()] {
-				return true
-			}
-		}
-		return false
-	}, cmp.Ignore())
-}
diff --git a/go/vt/vtgate/autocommit_test.go b/go/vt/vtgate/autocommit_test.go
index e0ebb8c670..1188cab480 100644
--- a/go/vt/vtgate/autocommit_test.go
+++ b/go/vt/vtgate/autocommit_test.go
@@ -424,7 +424,7 @@ func TestAutocommitDirectRangeTarget(t *testing.T) {
 		Autocommit:      true,
 		TransactionMode: vtgatepb.TransactionMode_MULTI,
 	}
-	sql := "delete from sharded_user_msgs limit 1000"
+	sql := "DELETE FROM sharded_user_msgs LIMIT 1000"
 
 	_, err := executor.Execute(context.Background(), "TestExecute", NewSafeSession(session), sql, map[string]*querypb.BindVariable{})
 	require.NoError(t, err)
diff --git a/go/vt/vtgate/engine/fake_vcursor_test.go b/go/vt/vtgate/engine/fake_vcursor_test.go
index c0d93370f4..7ff09281eb 100644
--- a/go/vt/vtgate/engine/fake_vcursor_test.go
+++ b/go/vt/vtgate/engine/fake_vcursor_test.go
@@ -58,11 +58,11 @@ func (t noopVCursor) SetContextTimeout(timeout time.Duration) context.CancelFunc
 func (t noopVCursor) RecordWarning(warning *querypb.QueryWarning) {
 }
 
-func (t noopVCursor) Execute(method string, query string, bindvars map[string]*querypb.BindVariable, rollbackOnError bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error) {
+func (t noopVCursor) Execute(method string, query string, bindvars map[string]*querypb.BindVariable, isDML bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error) {
 	panic("unimplemented")
 }
 
-func (t noopVCursor) ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, rollbackOnError, canAutocommit bool) (*sqltypes.Result, []error) {
+func (t noopVCursor) ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, isDML, autocommit bool) (*sqltypes.Result, []error) {
 	panic("unimplemented")
 }
 
@@ -78,7 +78,7 @@ func (t noopVCursor) StreamExecuteMulti(query string, rss []*srvtopo.ResolvedSha
 	panic("unimplemented")
 }
 
-func (t noopVCursor) ExecuteKeyspaceID(keyspace string, ksid []byte, query string, bindVars map[string]*querypb.BindVariable, rollbackOnError, autocommit bool) (*sqltypes.Result, error) {
+func (t noopVCursor) ExecuteKeyspaceID(keyspace string, ksid []byte, query string, bindVars map[string]*querypb.BindVariable, isDML, autocommit bool) (*sqltypes.Result, error) {
 	panic("unimplemented")
 }
 
@@ -121,7 +121,7 @@ func (f *loggingVCursor) RecordWarning(warning *querypb.QueryWarning) {
 	f.warnings = append(f.warnings, warning)
 }
 
-func (f *loggingVCursor) Execute(method string, query string, bindvars map[string]*querypb.BindVariable, rollbackOnError bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error) {
+func (f *loggingVCursor) Execute(method string, query string, bindvars map[string]*querypb.BindVariable, isDML bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error) {
 	name := "Unknown"
 	switch co {
 	case vtgatepb.CommitOrder_NORMAL:
@@ -133,12 +133,12 @@ func (f *loggingVCursor) Execute(method string, query string, bindvars map[strin
 	case vtgatepb.CommitOrder_AUTOCOMMIT:
 		name = "ExecuteAutocommit"
 	}
-	f.log = append(f.log, fmt.Sprintf("%s %s %v %v", name, query, printBindVars(bindvars), rollbackOnError))
+	f.log = append(f.log, fmt.Sprintf("%s %s %v %v", name, query, printBindVars(bindvars), isDML))
 	return f.nextResult()
 }
 
-func (f *loggingVCursor) ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, rollbackOnError, canAutocommit bool) (*sqltypes.Result, []error) {
-	f.log = append(f.log, fmt.Sprintf("ExecuteMultiShard %v%v %v", printResolvedShardQueries(rss, queries), rollbackOnError, canAutocommit))
+func (f *loggingVCursor) ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, isDML, canAutocommit bool) (*sqltypes.Result, []error) {
+	f.log = append(f.log, fmt.Sprintf("ExecuteMultiShard %v%v %v", printResolvedShardQueries(rss, queries), isDML, canAutocommit))
 	res, err := f.nextResult()
 	if err != nil {
 		return nil, []error{err}
@@ -202,8 +202,6 @@ func (f *loggingVCursor) ResolveDestinations(keyspace string, ids []*querypb.Val
 			shards = f.shards[:1]
 		case key.DestinationNone:
 			// Nothing to do here.
-		case key.DestinationShard:
-			shards = []string{destination.String()}
 		default:
 			return nil, nil, fmt.Errorf("unsupported destination: %v", destination)
 		}
diff --git a/go/vt/vtgate/engine/send_test.go b/go/vt/vtgate/engine/send_test.go
deleted file mode 100644
index 38d8ad510b..0000000000
--- a/go/vt/vtgate/engine/send_test.go
+++ /dev/null
@@ -1,112 +0,0 @@
-package engine
-
-import (
-	"errors"
-	"testing"
-
-	"github.com/stretchr/testify/require"
-
-	"vitess.io/vitess/go/vt/key"
-	querypb "vitess.io/vitess/go/vt/proto/query"
-	"vitess.io/vitess/go/vt/vtgate/vindexes"
-)
-
-func TestSendTable(t *testing.T) {
-	type testCase struct {
-		testName         string
-		sharded          bool
-		shards           []string
-		destination      key.Destination
-		expectedQueryLog []string
-		noAutoCommit     bool
-	}
-
-	singleShard := []string{"0"}
-	twoShards := []string{"-20", "20-"}
-	tests := []testCase{
-		{
-			testName:    "unsharded with no autocommit",
-			sharded:     false,
-			shards:      singleShard,
-			destination: key.DestinationAllShards{},
-			expectedQueryLog: []string{
-				`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
-				`ExecuteMultiShard ks.0: dummy_query {} false false`,
-			},
-			noAutoCommit: true,
-		},
-		{
-			testName:    "sharded with no autocommit",
-			sharded:     true,
-			shards:      twoShards,
-			destination: key.DestinationShard("20-"),
-			expectedQueryLog: []string{
-				`ResolveDestinations ks [] Destinations:DestinationShard(20-)`,
-				`ExecuteMultiShard ks.DestinationShard(20-): dummy_query {} false false`,
-			},
-			noAutoCommit: true,
-		},
-		{
-			testName:    "unsharded",
-			sharded:     false,
-			shards:      singleShard,
-			destination: key.DestinationAllShards{},
-			expectedQueryLog: []string{
-				`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
-				`ExecuteMultiShard ks.0: dummy_query {} true true`,
-			},
-			noAutoCommit: false,
-		},
-		{
-			testName:    "sharded with single shard destination",
-			sharded:     true,
-			shards:      twoShards,
-			destination: key.DestinationShard("20-"),
-			expectedQueryLog: []string{
-				`ResolveDestinations ks [] Destinations:DestinationShard(20-)`,
-				`ExecuteMultiShard ks.DestinationShard(20-): dummy_query {} true true`,
-			},
-			noAutoCommit: false,
-		},
-		{
-			testName:    "sharded with multi shard destination",
-			sharded:     true,
-			shards:      twoShards,
-			destination: key.DestinationAllShards{},
-			expectedQueryLog: []string{
-				`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
-				`ExecuteMultiShard ks.-20: dummy_query {} ks.20-: dummy_query {} true false`,
-			},
-			noAutoCommit: false,
-		},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.testName, func(t *testing.T) {
-			send := &Send{
-				Keyspace: &vindexes.Keyspace{
-					Name:    "ks",
-					Sharded: tc.sharded,
-				},
-				Query:             "dummy_query",
-				TargetDestination: tc.destination,
-				NoAutoCommit:      tc.noAutoCommit,
-			}
-			vc := &loggingVCursor{shards: tc.shards}
-			_, err := send.Execute(vc, map[string]*querypb.BindVariable{}, false)
-			require.NoError(t, err)
-			vc.ExpectLog(t, tc.expectedQueryLog)
-
-			// Failure cases
-			vc = &loggingVCursor{shardErr: errors.New("shard_error")}
-			_, err = send.Execute(vc, map[string]*querypb.BindVariable{}, false)
-			require.EqualError(t, err, "sendExecute: shard_error")
-
-			if !tc.sharded {
-				vc = &loggingVCursor{}
-				_, err = send.Execute(vc, map[string]*querypb.BindVariable{}, false)
-				require.EqualError(t, err, "Keyspace does not have exactly one shard: []")
-			}
-		})
-	}
-}
diff --git a/go/vt/vtgate/executor_dml_test.go b/go/vt/vtgate/executor_dml_test.go
index b0a47608df..d8f2a3cc62 100644
--- a/go/vt/vtgate/executor_dml_test.go
+++ b/go/vt/vtgate/executor_dml_test.go
@@ -21,9 +21,6 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/stretchr/testify/assert"
-	"vitess.io/vitess/go/test/utils"
-
 	"github.com/stretchr/testify/require"
 
 	"golang.org/x/net/context"
@@ -317,8 +314,12 @@ func TestUpdateNormalize(t *testing.T) {
 			"vtg2": sqltypes.TestBindVariable(int64(1)),
 		},
 	}}
-	assert.Empty(t, sbc1.Queries)
-	utils.MustMatch(t, sbc2.Queries, wantQueries, "didn't get expected queries")
+	if sbc1.Queries != nil {
+		t.Errorf("sbc1.Queries: %+v, want nil\n", sbc1.Queries)
+	}
+	if !reflect.DeepEqual(sbc2.Queries, wantQueries) {
+		t.Errorf("sbc2.Queries: %+v, want %+v\n", sbc2.Queries, wantQueries)
+	}
 	sbc2.Queries = nil
 	masterSession.TargetString = ""
 }
@@ -1523,102 +1524,186 @@ func TestInsertBadAutoInc(t *testing.T) {
 }
 
 func TestKeyDestRangeQuery(t *testing.T) {
+	executor, sbc1, sbc2, _ := createExecutorEnv()
+	// it works in a single shard key range
+	masterSession.TargetString = "TestExecutor[40-60]"
 
-	type testCase struct {
-		inputQuery, targetString string
-		expectedSbc1Query        string
-		expectedSbc2Query        string
-	}
-	deleteInput := "DELETE FROM sharded_user_msgs LIMIT 1000"
-	deleteOutput := "delete from sharded_user_msgs limit 1000"
-
-	selectInput := "SELECT * FROM sharded_user_msgs LIMIT 1"
-	selectOutput := "select * from sharded_user_msgs limit 1"
-	updateInput := "UPDATE sharded_user_msgs set message='test' LIMIT 1"
-	updateOutput := "update sharded_user_msgs set message = 'test' limit 1"
-	insertInput := "INSERT INTO sharded_user_msgs(message) VALUES('test')"
-	insertOutput := "insert into sharded_user_msgs(message) values ('test')"
-	tests := []testCase{
-		{
-			inputQuery:        deleteInput,
-			targetString:      "TestExecutor[-60]",
-			expectedSbc1Query: deleteOutput,
-			expectedSbc2Query: deleteOutput,
-		},
-		{
-			inputQuery:        deleteInput,
-			targetString:      "TestExecutor[40-60]",
-			expectedSbc2Query: deleteOutput,
-		},
-		{
-			inputQuery:        deleteInput,
-			targetString:      "TestExecutor[-]",
-			expectedSbc1Query: deleteOutput,
-			expectedSbc2Query: deleteOutput,
-		},
-		{
-			inputQuery:        selectInput,
-			targetString:      "TestExecutor[-]",
-			expectedSbc1Query: selectOutput,
-			expectedSbc2Query: selectOutput,
-		},
-		{
-			inputQuery:        updateInput,
-			targetString:      "TestExecutor[-]",
-			expectedSbc1Query: updateOutput,
-			expectedSbc2Query: updateOutput,
-		},
-		{
-			inputQuery:        insertInput,
-			targetString:      "TestExecutor:40-60",
-			expectedSbc2Query: insertOutput,
-		},
-		{
-			inputQuery:        insertInput,
-			targetString:      "TestExecutor:-20",
-			expectedSbc1Query: insertOutput,
-		},
+	_, err := executorExec(executor, "DELETE FROM sharded_user_msgs LIMIT 1000", nil)
+	require.NoError(t, err)
+	sql := "DELETE FROM sharded_user_msgs LIMIT 1000"
+	wantQueries := []*querypb.BoundQuery{{
+		Sql:           sql,
+		BindVariables: map[string]*querypb.BindVariable{},
+	}}
+
+	if len(sbc1.Queries) != 0 {
+		t.Errorf("sbc1.Queries: %+v, want %+v\n", sbc1.Queries, []*querypb.BoundQuery{})
 	}
+	testQueries(t, "sbc2", sbc2, wantQueries)
 
-	for _, tc := range tests {
-		t.Run(tc.targetString+" - "+tc.inputQuery, func(t *testing.T) {
-			executor, sbc1, sbc2, _ := createExecutorEnv()
+	sbc1.Queries = nil
+	sbc2.Queries = nil
 
-			masterSession.TargetString = tc.targetString
-			_, err := executorExec(executor, tc.inputQuery, nil)
-			require.NoError(t, err)
+	// it works with keyrange spanning two shards
+	masterSession.TargetString = "TestExecutor[-60]"
 
-			if tc.expectedSbc1Query == "" {
-				require.Empty(t, sbc1.BatchQueries, "sbc1")
-			} else {
-				assertQueriesContain(t, tc.expectedSbc1Query, "sbc1", sbc1)
-			}
+	_, err = executorExec(executor, sql, nil)
+	require.NoError(t, err)
+	testQueries(t, "sbc1", sbc1, wantQueries)
+	testQueries(t, "sbc1", sbc2, wantQueries)
 
-			if tc.expectedSbc2Query == "" {
-				require.Empty(t, sbc2.BatchQueries)
-			} else {
-				assertQueriesContain(t, tc.expectedSbc2Query, "sbc2", sbc2)
-			}
-		})
-	}
+	sbc1.Queries = nil
+	sbc2.Queries = nil
 
-	// it does not work for inserts
-	executor, _, _, _ := createExecutorEnv()
+	// it works with open ended key range
 	masterSession.TargetString = "TestExecutor[-]"
-	_, err := executorExec(executor, insertInput, nil)
 
-	require.EqualError(t, err, "range queries not supported for inserts: TestExecutor[-]")
+	_, err = executorExec(executor, sql, nil)
+	require.NoError(t, err)
+
+	testQueries(t, "sbc1", sbc1, wantQueries)
+	testQueries(t, "sbc2", sbc2, wantQueries)
+
+	sbc1.Queries = nil
+	sbc2.Queries = nil
+
+	// it works for select
+	sql = "SELECT * FROM sharded_user_msgs LIMIT 1"
+	wantQueries = []*querypb.BoundQuery{{
+		Sql:           sql,
+		BindVariables: map[string]*querypb.BindVariable{},
+	}}
+
+	_, err = executorExec(executor, sql, nil)
+	require.NoError(t, err)
+
+	testQueries(t, "sbc1", sbc1, wantQueries)
+	testQueries(t, "sbc2", sbc2, wantQueries)
+
+	sbc1.Queries = nil
+	sbc2.Queries = nil
+
+	// it works for updates
+	sql = "UPDATE sharded_user_msgs set message='test' LIMIT 1"
+
+	wantQueries = []*querypb.BoundQuery{{
+		Sql:           sql,
+		BindVariables: map[string]*querypb.BindVariable{},
+	}}
+
+	_, err = executorExec(executor, sql, nil)
+	require.NoError(t, err)
+
+	testQueries(t, "sbc1", sbc1, wantQueries)
+	testQueries(t, "sbc2", sbc2, wantQueries)
+
+	sbc1.Queries = nil
+	sbc2.Queries = nil
+
+	// it does not work for inserts
+	_, err = executorExec(executor, "INSERT INTO sharded_user_msgs(message) VALUES('test')", nil)
+
+	want := "range queries not supported for inserts: TestExecutor[-]"
+	if err == nil || err.Error() != want {
+		t.Errorf("got: %v, want %s", err, want)
+	}
 
+	sbc1.Queries = nil
+	sbc2.Queries = nil
 	masterSession.TargetString = ""
 }
 
-func assertQueriesContain(t *testing.T, sql, sbcName string, sbc *sandboxconn.SandboxConn) {
-	t.Helper()
-	expectedQuery := []*querypb.BoundQuery{{
+func TestKeyShardDestQuery(t *testing.T) {
+	executor, sbc1, sbc2, _ := createExecutorEnv()
+	// it works in a single shard key range
+	masterSession.TargetString = "TestExecutor:40-60"
+
+	_, err := executorExec(executor, "DELETE FROM sharded_user_msgs LIMIT 1000", nil)
+	require.NoError(t, err)
+	sql := "DELETE FROM sharded_user_msgs LIMIT 1000"
+	wantQueries := []*querypb.BoundQuery{{
+		Sql:           sql,
+		BindVariables: map[string]*querypb.BindVariable{},
+	}}
+
+	if len(sbc1.Queries) != 0 {
+		t.Errorf("sbc1.Queries: %+v, want %+v\n", sbc1.Queries, []*querypb.BoundQuery{})
+	}
+	testQueries(t, "sbc2", sbc2, wantQueries)
+
+	sbc1.Queries = nil
+	sbc2.Queries = nil
+
+	masterSession.TargetString = "TestExecutor:40-60"
+
+	_, err = executorExec(executor, sql, nil)
+	require.NoError(t, err)
+
+	testQueries(t, "sbc2", sbc2, wantQueries)
+
+	sbc1.Queries = nil
+	sbc2.Queries = nil
+
+	// it works for select
+	sql = "SELECT * FROM sharded_user_msgs LIMIT 1"
+	wantQueries = []*querypb.BoundQuery{{
+		Sql:           sql,
+		BindVariables: map[string]*querypb.BindVariable{},
+	}}
+
+	_, err = executorExec(executor, sql, nil)
+	require.NoError(t, err)
+
+	if len(sbc1.Queries) != 0 {
+		t.Errorf("sbc1.Queries: %+v, want %+v\n", sbc1.Queries, []*querypb.BoundQuery{})
+	}
+
+	testQueries(t, "sbc2", sbc2, wantQueries)
+
+	sbc1.Queries = nil
+	sbc2.Queries = nil
+
+	// it works for updates
+	sql = "UPDATE sharded_user_msgs set message='test' LIMIT 1"
+
+	wantQueries = []*querypb.BoundQuery{{
 		Sql:           sql,
 		BindVariables: map[string]*querypb.BindVariable{},
 	}}
-	testQueries(t, sbcName, sbc, expectedQuery)
+
+	_, err = executorExec(executor, sql, nil)
+
+	require.NoError(t, err)
+
+	if len(sbc1.Queries) != 0 {
+		t.Errorf("sbc1.Queries: %+v, want %+v\n", sbc1.Queries, []*querypb.BoundQuery{})
+	}
+
+	testQueries(t, "sbc2", sbc2, wantQueries)
+
+	sbc1.Queries = nil
+	sbc2.Queries = nil
+
+	// it works for inserts
+
+	sql = "INSERT INTO sharded_user_msgs(message) VALUES('test')"
+	_, err = executorExec(executor, sql, nil)
+
+	wantQueries = []*querypb.BoundQuery{{
+		Sql:           sql,
+		BindVariables: map[string]*querypb.BindVariable{},
+	}}
+	require.NoError(t, err)
+
+	if len(sbc1.Queries) != 0 {
+		t.Errorf("sbc1.Queries: %+v, want %+v\n", sbc1.Queries, []*querypb.BoundQuery{})
+	}
+
+	testQueries(t, "sbc2", sbc2, wantQueries)
+
+	sbc1.Queries = nil
+	sbc2.Queries = nil
+	masterSession.TargetString = ""
 }
 
 // Prepared statement tests
diff --git a/go/vt/vtgate/executor_select_test.go b/go/vt/vtgate/executor_select_test.go
index 6e53ec590d..502680c494 100644
--- a/go/vt/vtgate/executor_select_test.go
+++ b/go/vt/vtgate/executor_select_test.go
@@ -22,8 +22,6 @@ import (
 	"strings"
 	"testing"
 
-	"vitess.io/vitess/go/test/utils"
-
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -634,8 +632,12 @@ func TestSelectNormalize(t *testing.T) {
 			"vtg1": sqltypes.TestBindVariable(int64(1)),
 		},
 	}}
-	require.Empty(t, sbc1.Queries)
-	utils.MustMatch(t, sbc2.Queries, wantQueries, "sbc2.Queries")
+	if sbc1.Queries != nil {
+		t.Errorf("sbc1.Queries: %+v, want nil\n", sbc1.Queries)
+	}
+	if !reflect.DeepEqual(sbc2.Queries, wantQueries) {
+		t.Errorf("sbc2.Queries: %+v, want %+v\n", sbc2.Queries, wantQueries)
+	}
 	sbc2.Queries = nil
 	masterSession.TargetString = ""
 }
diff --git a/go/vt/vtgate/executor_test.go b/go/vt/vtgate/executor_test.go
index 938f434084..e9b8a67ac7 100644
--- a/go/vt/vtgate/executor_test.go
+++ b/go/vt/vtgate/executor_test.go
@@ -205,8 +205,9 @@ func TestDirectTargetRewrites(t *testing.T) {
 	}
 	sql := "select database()"
 
-	_, err := executor.Execute(context.Background(), "TestExecute", NewSafeSession(session), sql, map[string]*querypb.BindVariable{})
-	require.NoError(t, err)
+	if _, err := executor.Execute(context.Background(), "TestExecute", NewSafeSession(session), sql, map[string]*querypb.BindVariable{}); err != nil {
+		t.Error(err)
+	}
 	testQueries(t, "sbclookup", sbclookup, []*querypb.BoundQuery{{
 		Sql:           "select :__vtdbname as `database()` from dual",
 		BindVariables: map[string]*querypb.BindVariable{"__vtdbname": sqltypes.StringBindVariable("TestUnsharded")},
@@ -1958,8 +1959,8 @@ func TestVSchemaStats(t *testing.T) {
 
 func TestGetPlanUnnormalized(t *testing.T) {
 	r, _, _, _ := createExecutorEnv()
-	emptyvc, _ := newVCursorImpl(context.Background(), NewSafeSession(&vtgatepb.Session{TargetString: "@unknown"}), makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
-	unshardedvc, _ := newVCursorImpl(context.Background(), NewSafeSession(&vtgatepb.Session{TargetString: KsTestUnsharded + "@unknown"}), makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
+	emptyvc := newVCursorImpl(context.Background(), nil, "", 0, makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
+	unshardedvc := newVCursorImpl(context.Background(), nil, KsTestUnsharded, 0, makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
 
 	logStats1 := NewLogStats(context.Background(), "Test", "", nil)
 	query1 := "select * from music_user_map where id = 1"
@@ -2004,12 +2005,9 @@ func TestGetPlanUnnormalized(t *testing.T) {
 		KsTestUnsharded + "@unknown:" + query1,
 		"@unknown:" + query1,
 	}
-	if diff := cmp.Diff(want, r.plans.Keys()); diff != "" {
-		t.Errorf("\n-want,+got:\n%s", diff)
+	if keys := r.plans.Keys(); !reflect.DeepEqual(keys, want) {
+		t.Errorf("Plan keys: %s, want %s", keys, want)
 	}
-	//if keys := r.plans.Keys(); !reflect.DeepEqual(keys, want) {
-	//	t.Errorf("Plan keys: %s, want %s", keys, want)
-	//}
 	if logStats4.SQL != wantSQL {
 		t.Errorf("logstats sql want \"%s\" got \"%s\"", wantSQL, logStats4.SQL)
 	}
@@ -2017,7 +2015,7 @@ func TestGetPlanUnnormalized(t *testing.T) {
 
 func TestGetPlanCacheUnnormalized(t *testing.T) {
 	r, _, _, _ := createExecutorEnv()
-	emptyvc, _ := newVCursorImpl(context.Background(), NewSafeSession(&vtgatepb.Session{TargetString: "@unknown"}), makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
+	emptyvc := newVCursorImpl(context.Background(), nil, "", 0, makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
 	query1 := "select * from music_user_map where id = 1"
 	logStats1 := NewLogStats(context.Background(), "Test", "", nil)
 	_, err := r.getPlan(emptyvc, query1, makeComments(" /* comment */"), map[string]*querypb.BindVariable{}, true /* skipQueryPlanCache */, logStats1)
@@ -2042,7 +2040,7 @@ func TestGetPlanCacheUnnormalized(t *testing.T) {
 
 	// Skip cache using directive
 	r, _, _, _ = createExecutorEnv()
-	unshardedvc, _ := newVCursorImpl(context.Background(), NewSafeSession(&vtgatepb.Session{TargetString: KsTestUnsharded + "@unknown"}), makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
+	unshardedvc := newVCursorImpl(context.Background(), nil, KsTestUnsharded, 0, makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
 
 	query1 = "insert /*vt+ SKIP_QUERY_PLAN_CACHE=1 */ into user(id) values (1), (2)"
 	logStats1 = NewLogStats(context.Background(), "Test", "", nil)
@@ -2063,7 +2061,7 @@ func TestGetPlanCacheUnnormalized(t *testing.T) {
 func TestGetPlanCacheNormalized(t *testing.T) {
 	r, _, _, _ := createExecutorEnv()
 	r.normalize = true
-	emptyvc, _ := newVCursorImpl(context.Background(), NewSafeSession(&vtgatepb.Session{TargetString: "@unknown"}), makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
+	emptyvc := newVCursorImpl(context.Background(), nil, "", 0, makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
 	query1 := "select * from music_user_map where id = 1"
 	logStats1 := NewLogStats(context.Background(), "Test", "", nil)
 	_, err := r.getPlan(emptyvc, query1, makeComments(" /* comment */"), map[string]*querypb.BindVariable{}, true /* skipQueryPlanCache */, logStats1)
@@ -2088,7 +2086,7 @@ func TestGetPlanCacheNormalized(t *testing.T) {
 	// Skip cache using directive
 	r, _, _, _ = createExecutorEnv()
 	r.normalize = true
-	unshardedvc, _ := newVCursorImpl(context.Background(), NewSafeSession(&vtgatepb.Session{TargetString: KsTestUnsharded + "@unknown"}), makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
+	unshardedvc := newVCursorImpl(context.Background(), nil, KsTestUnsharded, 0, makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
 
 	query1 = "insert /*vt+ SKIP_QUERY_PLAN_CACHE=1 */ into user(id) values (1), (2)"
 	logStats1 = NewLogStats(context.Background(), "Test", "", nil)
@@ -2109,8 +2107,8 @@ func TestGetPlanCacheNormalized(t *testing.T) {
 func TestGetPlanNormalized(t *testing.T) {
 	r, _, _, _ := createExecutorEnv()
 	r.normalize = true
-	emptyvc, _ := newVCursorImpl(context.Background(), NewSafeSession(&vtgatepb.Session{TargetString: "@unknown"}), makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
-	unshardedvc, _ := newVCursorImpl(context.Background(), NewSafeSession(&vtgatepb.Session{TargetString: KsTestUnsharded + "@unknown"}), makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
+	emptyvc := newVCursorImpl(context.Background(), nil, "", 0, makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
+	unshardedvc := newVCursorImpl(context.Background(), nil, KsTestUnsharded, 0, makeComments(""), r, nil, r.VSchema(), r.resolver.resolver)
 
 	query1 := "select * from music_user_map where id = 1"
 	query2 := "select * from music_user_map where id = 2"
diff --git a/go/vt/vtgate/gateway_test_suite.go b/go/vt/vtgate/gateway_test_suite.go
index 6241399a21..db5c69bb62 100644
--- a/go/vt/vtgate/gateway_test_suite.go
+++ b/go/vt/vtgate/gateway_test_suite.go
@@ -14,9 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package vtgate
-
-// This file contains a test suite to run against a Gateway object.
+// Package gatewaytest contains a test suite to run against a Gateway object.
 // We re-use the tabletconn test suite, as it tests all queries and parameters
 // go through. There are two exceptions:
 // - the health check: we just make that one work, so the gateway knows the
@@ -24,6 +22,7 @@ package vtgate
 // - the error type returned: it's not a TabletError any more, but a ShardError.
 //   We still check the error code is correct though which is really all we care
 //   about.
+package vtgate
 
 import (
 	"testing"
@@ -80,7 +79,7 @@ func CreateFakeServers(t *testing.T) (*tabletconntest.FakeQueryService, *topo.Se
 // gatewayAdapter implements the TabletConn interface, but sends the
 // queries to the Gateway.
 type gatewayAdapter struct {
-	Gateway
+	*tabletGateway
 }
 
 // Close should be overridden to make sure we don't close the underlying Gateway.
@@ -92,12 +91,12 @@ func (ga *gatewayAdapter) Close(ctx context.Context) error {
 // gateway needs to be configured with one established connection for
 // tabletconntest.TestTarget.{Keyspace, Shard, TabletType} to the
 // provided tabletconntest.FakeQueryService.
-func TestSuite(t *testing.T, name string, g Gateway, f *tabletconntest.FakeQueryService) {
+func TestSuite(t *testing.T, name string, g *tabletGateway, f *tabletconntest.FakeQueryService) {
 
 	protocolName := "gateway-test-" + name
 
 	tabletconn.RegisterDialer(protocolName, func(tablet *topodatapb.Tablet, failFast grpcclient.FailFast) (queryservice.QueryService, error) {
-		return &gatewayAdapter{Gateway: g}, nil
+		return &gatewayAdapter{tabletGateway: g}, nil
 	})
 
 	tabletconntest.TestSuite(t, protocolName, &topodatapb.Tablet{
diff --git a/go/vt/vtgate/grpc_discovery_test.go b/go/vt/vtgate/grpc_discovery_test.go
index 03d22a810a..b6aa62633d 100644
--- a/go/vt/vtgate/grpc_discovery_test.go
+++ b/go/vt/vtgate/grpc_discovery_test.go
@@ -18,12 +18,11 @@ package vtgate
 
 import (
 	"flag"
+	"golang.org/x/net/context"
 	"net"
 	"testing"
 	"time"
 
-	"golang.org/x/net/context"
-
 	"google.golang.org/grpc"
 
 	"vitess.io/vitess/go/vt/discovery"
@@ -64,7 +63,7 @@ func TestGRPCDiscovery(t *testing.T) {
 	// Wait for the right tablets to be present.
 	hc := discovery.NewHealthCheck(10*time.Second, 2*time.Minute)
 	rs := srvtopo.NewResilientServer(ts, "TestGRPCDiscovery")
-	dg := NewDiscoveryGateway(context.Background(), hc, rs, cell, 2)
+	dg := NewTabletGateway(context.Background(), hc, rs, cell, 2)
 	hc.AddTablet(&topodatapb.Tablet{
 		Alias: &topodatapb.TabletAlias{
 			Cell: cell,
diff --git a/go/vt/vtgate/planbuilder/plan_test.go b/go/vt/vtgate/planbuilder/plan_test.go
index a73ed6c8c8..c4ac1d2006 100644
--- a/go/vt/vtgate/planbuilder/plan_test.go
+++ b/go/vt/vtgate/planbuilder/plan_test.go
@@ -142,10 +142,7 @@ func init() {
 }
 
 func TestPlan(t *testing.T) {
-	vschemaWrapper := &vschemaWrapper{
-		v: loadSchema(t, "schema_test.json"),
-	}
-
+	vschema := loadSchema(t, "schema_test.json")
 	testOutputTempDir, err := ioutil.TempDir("", "plan_test")
 	require.NoError(t, err)
 	// You will notice that some tests expect user.Id instead of user.id.
@@ -154,43 +151,24 @@ func TestPlan(t *testing.T) {
 	// the column is named as Id. This is to make sure that
 	// column names are case-preserved, but treated as
 	// case-insensitive even if they come from the vschema.
-	testFile(t, "aggr_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "dml_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "from_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "filter_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "postprocess_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "select_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "symtab_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "unsupported_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "vindex_func_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "wireup_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "memory_sort_cases.txt", testOutputTempDir, vschemaWrapper)
+	testFile(t, "aggr_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "dml_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "from_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "filter_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "postprocess_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "select_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "symtab_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "unsupported_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "vindex_func_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "wireup_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "memory_sort_cases.txt", testOutputTempDir, vschema)
 }
 
 func TestOne(t *testing.T) {
-	vschema := &vschemaWrapper{
-		v: loadSchema(t, "schema_test.json"),
-	}
-
+	vschema := loadSchema(t, "schema_test.json")
 	testFile(t, "onecase.txt", "", vschema)
 }
 
-func TestBypassPlanningFromFile(t *testing.T) {
-	testOutputTempDir, err := ioutil.TempDir("", "plan_test")
-	require.NoError(t, err)
-	vschema := &vschemaWrapper{
-		v: loadSchema(t, "schema_test.json"),
-		keyspace: &vindexes.Keyspace{
-			Name:    "main",
-			Sharded: false,
-		},
-		tabletType: topodatapb.TabletType_MASTER,
-		dest:       key.DestinationShard("-80"),
-	}
-
-	testFile(t, "bypass_cases.txt", testOutputTempDir, vschema)
-}
-
 func loadSchema(t *testing.T, filename string) *vindexes.VSchema {
 	formal, err := vindexes.LoadFormal(locateFile(filename))
 	if err != nil {
@@ -208,21 +186,8 @@ func loadSchema(t *testing.T, filename string) *vindexes.VSchema {
 	return vschema
 }
 
-var _ ContextVSchema = (*vschemaWrapper)(nil)
-
 type vschemaWrapper struct {
-	v          *vindexes.VSchema
-	keyspace   *vindexes.Keyspace
-	tabletType topodatapb.TabletType
-	dest       key.Destination
-}
-
-func (vw *vschemaWrapper) TabletType() topodatapb.TabletType {
-	return vw.tabletType
-}
-
-func (vw *vschemaWrapper) Destination() key.Destination {
-	return vw.dest
+	v *vindexes.VSchema
 }
 
 func (vw *vschemaWrapper) FindTable(tab sqlparser.TableName) (*vindexes.Table, string, topodatapb.TabletType, key.Destination, error) {
@@ -264,19 +229,21 @@ type testPlan struct {
 	Instructions engine.Primitive `json:",omitempty"`
 }
 
-func testFile(t *testing.T, filename, tempDir string, vschema *vschemaWrapper) {
+func testFile(t *testing.T, filename, tempDir string, vschema *vindexes.VSchema) {
 	t.Run(filename, func(t *testing.T) {
 		expected := &strings.Builder{}
 		fail := false
 		for tcase := range iterateExecFile(filename) {
 			t.Run(tcase.comments, func(t *testing.T) {
-				plan, err := Build(tcase.input, vschema)
+				plan, err := Build(tcase.input, &vschemaWrapper{
+					v: vschema,
+				})
 
 				out := getPlanOrErrorOutput(err, plan)
 
 				if out != tcase.output {
 					fail = true
-					t.Errorf("File: %s, Line: %v\n %s\n%s", filename, tcase.lineno, cmp.Diff(tcase.output, out), out)
+					t.Errorf("File: %s, Line: %v\n %s", filename, tcase.lineno, cmp.Diff(tcase.output, out))
 				}
 
 				if err != nil {
diff --git a/go/vt/vtgate/planbuilder/testdata/bypass_cases.txt b/go/vt/vtgate/planbuilder/testdata/bypass_cases.txt
deleted file mode 100644
index e68e934d78..0000000000
--- a/go/vt/vtgate/planbuilder/testdata/bypass_cases.txt
+++ /dev/null
@@ -1,63 +0,0 @@
-# select bypass
-"select count(*), col from unsharded"
-{
-  "Original": "select count(*), col from unsharded",
-  "Instructions": {
-    "Opcode": "Send",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "TargetDestination": "-80",
-    "Query": "select count(*), col from unsharded",
-    "NoAutoCommit": true
-  }
-}
-
-# update bypass
-"update user set val = 1 where id = 18446744073709551616 and id = 1"
-{
-  "Original": "update user set val = 1 where id = 18446744073709551616 and id = 1",
-  "Instructions": {
-    "Opcode": "Send",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "TargetDestination": "-80",
-    "Query": "update user set val = 1 where id = 18446744073709551616 and id = 1",
-    "NoAutoCommit": false
-  }
-}
-
-# delete bypass
-"DELETE FROM USER WHERE ID = 42"
-{
-  "Original": "DELETE FROM USER WHERE ID = 42",
-  "Instructions": {
-    "Opcode": "Send",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "TargetDestination": "-80",
-    "Query": "delete from USER where ID = 42",
-    "NoAutoCommit": false
-  }
-}
-
-# insert bypass
-"INSERT INTO USER (ID, NAME) VALUES (42, 'ms X')"
-{
-  "Original": "INSERT INTO USER (ID, NAME) VALUES (42, 'ms X')",
-  "Instructions": {
-    "Opcode": "Send",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "TargetDestination": "-80",
-    "Query": "insert into USER(ID, NAME) values (42, 'ms X')",
-    "NoAutoCommit": false
-  }
-}
diff --git a/go/vt/vtgate/scatter_conn_test.go b/go/vt/vtgate/scatter_conn_test.go
index ede85cc5cf..6267272e2e 100644
--- a/go/vt/vtgate/scatter_conn_test.go
+++ b/go/vt/vtgate/scatter_conn_test.go
@@ -652,7 +652,7 @@ func newTestScatterConn(hc discovery.HealthCheck, serv srvtopo.Server, cell stri
 	// The topo.Server is used to start watching the cells described
 	// in '-cells_to_watch' command line parameter, which is
 	// empty by default. So it's unused in this test, set to nil.
-	gw := GatewayCreator()(context.Background(), hc, serv, cell, 3)
+	gw := NewTabletGateway(context.Background(), hc, serv, cell, 3)
 	tc := NewTxConn(gw, vtgatepb.TransactionMode_TWOPC)
 	return NewScatterConn("", tc, gw, hc)
 }
diff --git a/go/vt/vtgate/status_test.go b/go/vt/vtgate/status_test.go
index 3c420e68ba..8472faae5d 100644
--- a/go/vt/vtgate/status_test.go
+++ b/go/vt/vtgate/status_test.go
@@ -22,7 +22,6 @@ import (
 	"reflect"
 	"testing"
 	"time"
-
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 )
 
diff --git a/go/vt/vtgate/discoverygateway_test.go b/go/vt/vtgate/tabletgateway_test.go
similarity index 90%
rename from go/vt/vtgate/discoverygateway_test.go
rename to go/vt/vtgate/tabletgateway_test.go
index ce7508f96e..ed231fbbee 100644
--- a/go/vt/vtgate/discoverygateway_test.go
+++ b/go/vt/vtgate/tabletgateway_test.go
@@ -18,11 +18,10 @@ package vtgate
 
 import (
 	"fmt"
+	"golang.org/x/net/context"
 	"strings"
 	"testing"
 
-	"golang.org/x/net/context"
-
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/discovery"
 	"vitess.io/vitess/go/vt/srvtopo/srvtopotest"
@@ -37,23 +36,23 @@ import (
 )
 
 func TestDiscoveryGatewayExecute(t *testing.T) {
-	testDiscoveryGatewayGeneric(t, func(dg *discoveryGateway, target *querypb.Target) error {
+	testDiscoveryGatewayGeneric(t, func(dg *tabletGateway, target *querypb.Target) error {
 		_, err := dg.Execute(context.Background(), target, "query", nil, 0, nil)
 		return err
 	})
-	testDiscoveryGatewayTransact(t, func(dg *discoveryGateway, target *querypb.Target) error {
+	testDiscoveryGatewayTransact(t, func(dg *tabletGateway, target *querypb.Target) error {
 		_, err := dg.Execute(context.Background(), target, "query", nil, 1, nil)
 		return err
 	})
 }
 
 func TestDiscoveryGatewayExecuteBatch(t *testing.T) {
-	testDiscoveryGatewayGeneric(t, func(dg *discoveryGateway, target *querypb.Target) error {
+	testDiscoveryGatewayGeneric(t, func(dg *tabletGateway, target *querypb.Target) error {
 		queries := []*querypb.BoundQuery{{Sql: "query", BindVariables: nil}}
 		_, err := dg.ExecuteBatch(context.Background(), target, queries, false, 0, nil)
 		return err
 	})
-	testDiscoveryGatewayTransact(t, func(dg *discoveryGateway, target *querypb.Target) error {
+	testDiscoveryGatewayTransact(t, func(dg *tabletGateway, target *querypb.Target) error {
 		queries := []*querypb.BoundQuery{{Sql: "query", BindVariables: nil}}
 		_, err := dg.ExecuteBatch(context.Background(), target, queries, false, 1, nil)
 		return err
@@ -61,7 +60,7 @@ func TestDiscoveryGatewayExecuteBatch(t *testing.T) {
 }
 
 func TestDiscoveryGatewayExecuteStream(t *testing.T) {
-	testDiscoveryGatewayGeneric(t, func(dg *discoveryGateway, target *querypb.Target) error {
+	testDiscoveryGatewayGeneric(t, func(dg *tabletGateway, target *querypb.Target) error {
 		err := dg.StreamExecute(context.Background(), target, "query", nil, 0, nil, func(qr *sqltypes.Result) error {
 			return nil
 		})
@@ -70,33 +69,33 @@ func TestDiscoveryGatewayExecuteStream(t *testing.T) {
 }
 
 func TestDiscoveryGatewayBegin(t *testing.T) {
-	testDiscoveryGatewayGeneric(t, func(dg *discoveryGateway, target *querypb.Target) error {
+	testDiscoveryGatewayGeneric(t, func(dg *tabletGateway, target *querypb.Target) error {
 		_, err := dg.Begin(context.Background(), target, nil)
 		return err
 	})
 }
 
 func TestDiscoveryGatewayCommit(t *testing.T) {
-	testDiscoveryGatewayTransact(t, func(dg *discoveryGateway, target *querypb.Target) error {
+	testDiscoveryGatewayTransact(t, func(dg *tabletGateway, target *querypb.Target) error {
 		return dg.Commit(context.Background(), target, 1)
 	})
 }
 
 func TestDiscoveryGatewayRollback(t *testing.T) {
-	testDiscoveryGatewayTransact(t, func(dg *discoveryGateway, target *querypb.Target) error {
+	testDiscoveryGatewayTransact(t, func(dg *tabletGateway, target *querypb.Target) error {
 		return dg.Rollback(context.Background(), target, 1)
 	})
 }
 
 func TestDiscoveryGatewayBeginExecute(t *testing.T) {
-	testDiscoveryGatewayGeneric(t, func(dg *discoveryGateway, target *querypb.Target) error {
+	testDiscoveryGatewayGeneric(t, func(dg *tabletGateway, target *querypb.Target) error {
 		_, _, err := dg.BeginExecute(context.Background(), target, "query", nil, nil)
 		return err
 	})
 }
 
 func TestDiscoveryGatewayBeginExecuteBatch(t *testing.T) {
-	testDiscoveryGatewayGeneric(t, func(dg *discoveryGateway, target *querypb.Target) error {
+	testDiscoveryGatewayGeneric(t, func(dg *tabletGateway, target *querypb.Target) error {
 		queries := []*querypb.BoundQuery{{Sql: "query", BindVariables: nil}}
 		_, _, err := dg.BeginExecuteBatch(context.Background(), target, queries, false, nil)
 		return err
@@ -107,7 +106,7 @@ func TestDiscoveryGatewayGetTablets(t *testing.T) {
 	keyspace := "ks"
 	shard := "0"
 	hc := discovery.NewFakeHealthCheck()
-	dg := NewDiscoveryGateway(context.Background(), hc, nil, "local", 2)
+	dg := NewTabletGateway(context.Background(), hc, nil, "local", 2)
 
 	// replica should only use local ones
 	hc.Reset()
@@ -215,7 +214,7 @@ func TestDiscoveryGatewayGetTabletsInRegion(t *testing.T) {
 		Cells: []string{"local-west", "local-east"},
 	}
 
-	dg := NewDiscoveryGateway(context.Background(), hc, srvTopo, "local-west", 2)
+	dg := NewTabletGateway(context.Background(), hc, srvTopo, "local-west", 2)
 
 	ts.CreateCellsAlias(context.Background(), "local", cellsAlias)
 
@@ -245,7 +244,7 @@ func TestDiscoveryGatewayGetTabletsWithRegion(t *testing.T) {
 		Cells: []string{"local-west", "local-east"},
 	}
 
-	dg := NewDiscoveryGateway(context.Background(), hc, srvTopo, "local", 2)
+	dg := NewTabletGateway(context.Background(), hc, srvTopo, "local", 2)
 
 	ts.CreateCellsAlias(context.Background(), "local", cellsAlias)
 
@@ -264,7 +263,7 @@ func TestDiscoveryGatewayGetTabletsWithRegion(t *testing.T) {
 	}
 }
 
-func testDiscoveryGatewayGeneric(t *testing.T, f func(dg *discoveryGateway, target *querypb.Target) error) {
+func testDiscoveryGatewayGeneric(t *testing.T, f func(dg *tabletGateway, target *querypb.Target) error) {
 	keyspace := "ks"
 	shard := "0"
 	tabletType := topodatapb.TabletType_REPLICA
@@ -274,7 +273,7 @@ func testDiscoveryGatewayGeneric(t *testing.T, f func(dg *discoveryGateway, targ
 		TabletType: tabletType,
 	}
 	hc := discovery.NewFakeHealthCheck()
-	dg := NewDiscoveryGateway(context.Background(), hc, nil, "cell", 2)
+	dg := NewTabletGateway(context.Background(), hc, nil, "cell", 2)
 
 	// no tablet
 	hc.Reset()
@@ -347,7 +346,7 @@ func testDiscoveryGatewayGeneric(t *testing.T, f func(dg *discoveryGateway, targ
 	}
 }
 
-func testDiscoveryGatewayTransact(t *testing.T, f func(dg *discoveryGateway, target *querypb.Target) error) {
+func testDiscoveryGatewayTransact(t *testing.T, f func(dg *tabletGateway, target *querypb.Target) error) {
 	keyspace := "ks"
 	shard := "0"
 	tabletType := topodatapb.TabletType_REPLICA
@@ -357,7 +356,7 @@ func testDiscoveryGatewayTransact(t *testing.T, f func(dg *discoveryGateway, tar
 		TabletType: tabletType,
 	}
 	hc := discovery.NewFakeHealthCheck()
-	dg := NewDiscoveryGateway(context.Background(), hc, nil, "cell", 2)
+	dg := NewTabletGateway(context.Background(), hc, nil, "cell", 2)
 
 	// retry error - no retry
 	hc.Reset()
diff --git a/go/vt/vtgate/vcursor_impl_test.go b/go/vt/vtgate/vcursor_impl_test.go
deleted file mode 100644
index d6062dca44..0000000000
--- a/go/vt/vtgate/vcursor_impl_test.go
+++ /dev/null
@@ -1,132 +0,0 @@
-package vtgate
-
-import (
-	"context"
-	"testing"
-
-	"vitess.io/vitess/go/vt/key"
-	"vitess.io/vitess/go/vt/vtgate/vindexes"
-
-	"github.com/stretchr/testify/require"
-	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
-	vtgatepb "vitess.io/vitess/go/vt/proto/vtgate"
-	"vitess.io/vitess/go/vt/sqlparser"
-)
-
-func TestDestinationKeyspace(t *testing.T) {
-	ks1 := &vindexes.Keyspace{
-		Name:    "ks1",
-		Sharded: false,
-	}
-	ks1Schema := &vindexes.KeyspaceSchema{
-		Keyspace: ks1,
-		Tables:   nil,
-		Vindexes: nil,
-		Error:    nil,
-	}
-	ks2 := &vindexes.Keyspace{
-		Name:    "ks2",
-		Sharded: false,
-	}
-	ks2Schema := &vindexes.KeyspaceSchema{
-		Keyspace: ks2,
-		Tables:   nil,
-		Vindexes: nil,
-		Error:    nil,
-	}
-	vschemaWith2KS := &vindexes.VSchema{
-		Keyspaces: map[string]*vindexes.KeyspaceSchema{
-			ks1.Name: ks1Schema,
-			ks2.Name: ks2Schema,
-		}}
-
-	vschemaWith1KS := &vindexes.VSchema{
-		Keyspaces: map[string]*vindexes.KeyspaceSchema{
-			ks1.Name: ks1Schema,
-		}}
-
-	type testCase struct {
-		vschema                 *vindexes.VSchema
-		targetString, qualifier string
-		expectedError           string
-		expectedKeyspace        string
-		expectedDest            key.Destination
-		expectedTabletType      topodatapb.TabletType
-	}
-
-	tests := []testCase{{
-		vschema:            vschemaWith1KS,
-		targetString:       "",
-		qualifier:          "",
-		expectedKeyspace:   ks1.Name,
-		expectedDest:       nil,
-		expectedTabletType: topodatapb.TabletType_MASTER,
-	}, {
-		vschema:            vschemaWith1KS,
-		targetString:       "ks1",
-		qualifier:          "",
-		expectedKeyspace:   ks1.Name,
-		expectedDest:       nil,
-		expectedTabletType: topodatapb.TabletType_MASTER,
-	}, {
-		vschema:            vschemaWith1KS,
-		targetString:       "ks1:-80",
-		qualifier:          "",
-		expectedKeyspace:   ks1.Name,
-		expectedDest:       key.DestinationShard("-80"),
-		expectedTabletType: topodatapb.TabletType_MASTER,
-	}, {
-		vschema:            vschemaWith1KS,
-		targetString:       "ks1@replica",
-		qualifier:          "",
-		expectedKeyspace:   ks1.Name,
-		expectedDest:       nil,
-		expectedTabletType: topodatapb.TabletType_REPLICA,
-	}, {
-		vschema:            vschemaWith1KS,
-		targetString:       "ks1:-80@replica",
-		qualifier:          "",
-		expectedKeyspace:   ks1.Name,
-		expectedDest:       key.DestinationShard("-80"),
-		expectedTabletType: topodatapb.TabletType_REPLICA,
-	}, {
-		vschema:            vschemaWith1KS,
-		targetString:       "",
-		qualifier:          "ks1",
-		expectedKeyspace:   ks1.Name,
-		expectedDest:       nil,
-		expectedTabletType: topodatapb.TabletType_MASTER,
-	}, {
-		vschema:       vschemaWith1KS,
-		targetString:  "ks2",
-		qualifier:     "",
-		expectedError: "no keyspace with name [ks2] found",
-	}, {
-		vschema:       vschemaWith1KS,
-		targetString:  "ks2:-80",
-		qualifier:     "",
-		expectedError: "no keyspace with name [ks2] found",
-	}, {
-		vschema:       vschemaWith1KS,
-		targetString:  "",
-		qualifier:     "ks2",
-		expectedError: "no keyspace with name [ks2] found",
-	}, {
-		vschema:       vschemaWith2KS,
-		targetString:  "",
-		expectedError: "keyspace not specified",
-	}}
-
-	for _, tc := range tests {
-		impl, _ := newVCursorImpl(context.Background(), NewSafeSession(&vtgatepb.Session{TargetString: tc.targetString}), sqlparser.MarginComments{}, nil, nil, tc.vschema, nil)
-		dest, keyspace, tabletType, err := impl.TargetDestination(tc.qualifier)
-		if tc.expectedError == "" {
-			require.NoError(t, err)
-			require.Equal(t, tc.expectedDest, dest)
-			require.Equal(t, tc.expectedKeyspace, keyspace.Name)
-			require.Equal(t, tc.expectedTabletType, tabletType)
-		} else {
-			require.EqualError(t, err, tc.expectedError)
-		}
-	}
-}
diff --git a/go/vt/vtgate/vindexes/consistent_lookup_test.go b/go/vt/vtgate/vindexes/consistent_lookup_test.go
index 346063def6..dbb95dbc4d 100644
--- a/go/vt/vtgate/vindexes/consistent_lookup_test.go
+++ b/go/vt/vtgate/vindexes/consistent_lookup_test.go
@@ -522,7 +522,7 @@ func (vc *loggingVCursor) AddResult(qr *sqltypes.Result, err error) {
 	vc.errors = append(vc.errors, err)
 }
 
-func (vc *loggingVCursor) Execute(method string, query string, bindvars map[string]*querypb.BindVariable, rollbackOnError bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error) {
+func (vc *loggingVCursor) Execute(method string, query string, bindvars map[string]*querypb.BindVariable, isDML bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error) {
 	name := "Unknown"
 	switch co {
 	case vtgatepb.CommitOrder_NORMAL:
@@ -534,14 +534,14 @@ func (vc *loggingVCursor) Execute(method string, query string, bindvars map[stri
 	case vtgatepb.CommitOrder_AUTOCOMMIT:
 		name = "ExecuteAutocommit"
 	}
-	return vc.execute(name, query, bindvars, rollbackOnError)
+	return vc.execute(name, query, bindvars, isDML)
 }
 
-func (vc *loggingVCursor) ExecuteKeyspaceID(keyspace string, ksid []byte, query string, bindVars map[string]*querypb.BindVariable, rollbackOnError, autocommit bool) (*sqltypes.Result, error) {
-	return vc.execute("ExecuteKeyspaceID", query, bindVars, rollbackOnError)
+func (vc *loggingVCursor) ExecuteKeyspaceID(keyspace string, ksid []byte, query string, bindVars map[string]*querypb.BindVariable, isDML, autocommit bool) (*sqltypes.Result, error) {
+	return vc.execute("ExecuteKeyspaceID", query, bindVars, isDML)
 }
 
-func (vc *loggingVCursor) execute(method string, query string, bindvars map[string]*querypb.BindVariable, rollbackOnError bool) (*sqltypes.Result, error) {
+func (vc *loggingVCursor) execute(method string, query string, bindvars map[string]*querypb.BindVariable, isDML bool) (*sqltypes.Result, error) {
 	if vc.index >= len(vc.results) {
 		return nil, fmt.Errorf("ran out of results to return: %s", query)
 	}
@@ -550,7 +550,7 @@ func (vc *loggingVCursor) execute(method string, query string, bindvars map[stri
 		bvl = append(bvl, bv{Name: k, Bv: string(v.Value)})
 	}
 	sort.Slice(bvl, func(i, j int) bool { return bvl[i].Name < bvl[j].Name })
-	vc.log = append(vc.log, fmt.Sprintf("%s %s %v %v", method, query, bvl, rollbackOnError))
+	vc.log = append(vc.log, fmt.Sprintf("%s %s %v %v", method, query, bvl, isDML))
 	idx := vc.index
 	vc.index++
 	if vc.errors[idx] != nil {
diff --git a/go/vt/vtgate/vindexes/lookup_test.go b/go/vt/vtgate/vindexes/lookup_test.go
index 827e0708dd..961a63725a 100644
--- a/go/vt/vtgate/vindexes/lookup_test.go
+++ b/go/vt/vtgate/vindexes/lookup_test.go
@@ -45,7 +45,7 @@ type vcursor struct {
 	pre, post   int
 }
 
-func (vc *vcursor) Execute(method string, query string, bindvars map[string]*querypb.BindVariable, rollbackOnError bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error) {
+func (vc *vcursor) Execute(method string, query string, bindvars map[string]*querypb.BindVariable, isDML bool, co vtgatepb.CommitOrder) (*sqltypes.Result, error) {
 	switch co {
 	case vtgatepb.CommitOrder_PRE:
 		vc.pre++
@@ -54,14 +54,14 @@ func (vc *vcursor) Execute(method string, query string, bindvars map[string]*que
 	case vtgatepb.CommitOrder_AUTOCOMMIT:
 		vc.autocommits++
 	}
-	return vc.execute(method, query, bindvars, rollbackOnError)
+	return vc.execute(method, query, bindvars, isDML)
 }
 
-func (vc *vcursor) ExecuteKeyspaceID(keyspace string, ksid []byte, query string, bindVars map[string]*querypb.BindVariable, rollbackOnError, autocommit bool) (*sqltypes.Result, error) {
-	return vc.execute("ExecuteKeyspaceID", query, bindVars, rollbackOnError)
+func (vc *vcursor) ExecuteKeyspaceID(keyspace string, ksid []byte, query string, bindVars map[string]*querypb.BindVariable, isDML, autocommit bool) (*sqltypes.Result, error) {
+	return vc.execute("ExecuteKeyspaceID", query, bindVars, isDML)
 }
 
-func (vc *vcursor) execute(method string, query string, bindvars map[string]*querypb.BindVariable, rollbackOnError bool) (*sqltypes.Result, error) {
+func (vc *vcursor) execute(method string, query string, bindvars map[string]*querypb.BindVariable, isDML bool) (*sqltypes.Result, error) {
 	vc.queries = append(vc.queries, &querypb.BoundQuery{
 		Sql:           query,
 		BindVariables: bindvars,
diff --git a/go/vt/vtgate/vstream_manager_test.go b/go/vt/vtgate/vstream_manager_test.go
index 94aeea2a1e..44eead7bea 100644
--- a/go/vt/vtgate/vstream_manager_test.go
+++ b/go/vt/vtgate/vstream_manager_test.go
@@ -873,7 +873,7 @@ func TestResolveVStreamParams(t *testing.T) {
 }
 
 func newTestVStreamManager(hc discovery.HealthCheck, serv srvtopo.Server, cell string) *vstreamManager {
-	gw := NewDiscoveryGateway(context.Background(), hc, serv, cell, 3)
+	gw := NewTabletGateway(context.Background(), hc, serv, cell, 3)
 	srvResolver := srvtopo.NewResolver(serv, gw, cell)
 	return newVStreamManager(srvResolver, serv, cell)
 }
diff --git a/go/vt/vttablet/tabletmanager/vreplication/framework_test.go b/go/vt/vttablet/tabletmanager/vreplication/framework_test.go
index 72b9e2c3b8..a87f6fe0df 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/framework_test.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/framework_test.go
@@ -497,7 +497,7 @@ func expectNontxQueries(t *testing.T, queries []string) {
 	retry:
 		select {
 		case got = <-globalDBQueries:
-			if got == "begin" || got == "commit" || strings.Contains(got, "_vt.vreplication") {
+			if got == "begin" || got == "commit" || got == "rollback" || strings.Contains(got, "_vt.vreplication") {
 				goto retry
 			}
 			var match bool
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vcopier_flaky_test.go b/go/vt/vttablet/tabletmanager/vreplication/vcopier_test.go
similarity index 97%
rename from go/vt/vttablet/tabletmanager/vreplication/vcopier_flaky_test.go
rename to go/vt/vttablet/tabletmanager/vreplication/vcopier_test.go
index 9ee4648750..b698c93d86 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vcopier_flaky_test.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vcopier_test.go
@@ -190,43 +190,28 @@ func TestPlayerCopyBigTable(t *testing.T) {
 		expectDeleteQueries(t)
 	}()
 
-	expectDBClientQueries(t, []string{
-		"/insert into _vt.vreplication",
+	expectNontxQueries(t, []string{
 		// Create the list of tables to copy and transition to Copying state.
-		"begin",
 		"/insert into _vt.copy_state",
-		"/update _vt.vreplication set state='Copying'",
-		"commit",
 		// The first fast-forward has no starting point. So, it just saves the current position.
-		"/update _vt.vreplication set pos=",
-		"begin",
 		"insert into dst(id,val) values (1,'aaa')",
 		`/update _vt.copy_state set lastpk='fields:<name:\\"id\\" type:INT32 > rows:<lengths:1 values:\\"1\\" > ' where vrepl_id=.*`,
-		"commit",
 		// The next catchup executes the new row insert, but will be a no-op.
-		"begin",
 		"insert into dst(id,val) select 3, 'ccc' from dual where (3) <= (1)",
-		"/update _vt.vreplication set pos=",
-		"commit",
 		// fastForward has nothing to add. Just saves position.
-		"begin",
-		"/update _vt.vreplication set pos=",
-		"commit",
 		// Second row gets copied.
-		"begin",
 		"insert into dst(id,val) values (2,'bbb')",
 		`/update _vt.copy_state set lastpk='fields:<name:\\"id\\" type:INT32 > rows:<lengths:1 values:\\"2\\" > ' where vrepl_id=.*`,
-		"commit",
 		// Third row copied without going back to catchup state.
-		"begin",
 		"insert into dst(id,val) values (3,'ccc')",
 		`/update _vt.copy_state set lastpk='fields:<name:\\"id\\" type:INT32 > rows:<lengths:1 values:\\"3\\" > ' where vrepl_id=.*`,
-		"commit",
 		"/delete from _vt.copy_state.*dst",
 		// Copy is done. Go into running state.
-		"/update _vt.vreplication set state='Running'",
 		// All tables copied. Final catch up followed by Running state.
 	})
+	expectDBClientQueries(t, []string{
+		"/update _vt.vreplication set state='Running'",
+	})
 	expectData(t, "dst", [][]string{
 		{"1", "aaa"},
 		{"2", "bbb"},
@@ -317,43 +302,28 @@ func TestPlayerCopyWildcardRule(t *testing.T) {
 		expectDeleteQueries(t)
 	}()
 
-	expectDBClientQueries(t, []string{
-		"/insert into _vt.vreplication",
+	expectNontxQueries(t, []string{
 		// Create the list of tables to copy and transition to Copying state.
-		"begin",
 		"/insert into _vt.copy_state",
-		"/update _vt.vreplication set state='Copying'",
-		"commit",
 		// The first fast-forward has no starting point. So, it just saves the current position.
-		"/update _vt.vreplication set pos=",
-		"begin",
 		"insert into src(id,val) values (1,'aaa')",
 		`/update _vt.copy_state set lastpk='fields:<name:\\"id\\" type:INT32 > rows:<lengths:1 values:\\"1\\" > ' where vrepl_id=.*`,
-		"commit",
 		// The next catchup executes the new row insert, but will be a no-op.
-		"begin",
 		"insert into src(id,val) select 3, 'ccc' from dual where (3) <= (1)",
-		"/update _vt.vreplication set pos=",
-		"commit",
 		// fastForward has nothing to add. Just saves position.
-		"begin",
-		"/update _vt.vreplication set pos=",
-		"commit",
 		// Second row gets copied.
-		"begin",
 		"insert into src(id,val) values (2,'bbb')",
 		`/update _vt.copy_state set lastpk='fields:<name:\\"id\\" type:INT32 > rows:<lengths:1 values:\\"2\\" > ' where vrepl_id=.*`,
-		"commit",
 		// Third row copied without going back to catchup state.
-		"begin",
 		"insert into src(id,val) values (3,'ccc')",
 		`/update _vt.copy_state set lastpk='fields:<name:\\"id\\" type:INT32 > rows:<lengths:1 values:\\"3\\" > ' where vrepl_id=.*`,
-		"commit",
 		"/delete from _vt.copy_state.*src",
 		// Copy is done. Go into running state.
-		"/update _vt.vreplication set state='Running'",
 		// All tables copied. Final catch up followed by Running state.
 	})
+	expectDBClientQueries(t, []string{
+		"/update _vt.vreplication set state='Running'",
+	})
 	expectData(t, "src", [][]string{
 		{"1", "aaa"},
 		{"2", "bbb"},
diff --git a/go/vt/vttablet/tabletserver/messager/message_manager_test.go b/go/vt/vttablet/tabletserver/messager/message_manager_test.go
index c7b41a3e63..1cee1b698f 100644
--- a/go/vt/vttablet/tabletserver/messager/message_manager_test.go
+++ b/go/vt/vttablet/tabletserver/messager/message_manager_test.go
@@ -65,24 +65,6 @@ func newMMTable() *schema.Table {
 			Fields:             testFields,
 			AckWaitDuration:    1 * time.Second,
 			PurgeAfterDuration: 3 * time.Second,
-			MinBackoff:         1 * time.Second,
-			BatchSize:          1,
-			CacheSize:          10,
-			PollInterval:       1 * time.Second,
-		},
-	}
-}
-
-func newMMTableWithBackoff() *schema.Table {
-	return &schema.Table{
-		Name: sqlparser.NewTableIdent("foo"),
-		Type: schema.Message,
-		MessageInfo: &schema.MessageInfo{
-			Fields:             testFields,
-			AckWaitDuration:    10 * time.Second,
-			PurgeAfterDuration: 3 * time.Second,
-			MinBackoff:         1 * time.Second,
-			MaxBackoff:         4 * time.Second,
 			BatchSize:          1,
 			CacheSize:          10,
 			PollInterval:       1 * time.Second,
@@ -750,7 +732,7 @@ func TestMMGenerate(t *testing.T) {
 	}
 
 	query, bv = mm.GeneratePostponeQuery([]string{"1", "2"})
-	wantQuery = "update foo set time_next = :time_now+(:min_backoff<<ifnull(epoch, 0)), epoch = ifnull(epoch, 0)+1 where id in ::ids and time_acked is null"
+	wantQuery = "update foo set time_next = :time_now+(:wait_time<<ifnull(epoch, 0)), epoch = ifnull(epoch, 0)+1 where id in ::ids and time_acked is null"
 	if query != wantQuery {
 		t.Errorf("GeneratePostponeQuery query: %s, want %s", query, wantQuery)
 	}
@@ -761,8 +743,8 @@ func TestMMGenerate(t *testing.T) {
 		delete(bv, "time_now")
 	}
 	wantbv := map[string]*querypb.BindVariable{
-		"min_backoff": sqltypes.Int64BindVariable(1e9),
-		"ids":         wantids,
+		"wait_time": sqltypes.Int64BindVariable(1e9),
+		"ids":       wantids,
 	}
 	if !reflect.DeepEqual(bv, wantbv) {
 		t.Errorf("gotid: %v, want %v", bv, wantbv)
@@ -781,34 +763,6 @@ func TestMMGenerate(t *testing.T) {
 	}
 }
 
-func TestMMGenerateWithBackoff(t *testing.T) {
-	mm := newMessageManager(newFakeTabletServer(), newFakeVStreamer(), newMMTableWithBackoff(), sync2.NewSemaphore(1, 0))
-	mm.Open()
-	defer mm.Close()
-
-	wantids := sqltypes.TestBindVariable([]interface{}{"1", "2"})
-
-	query, bv := mm.GeneratePostponeQuery([]string{"1", "2"})
-	wantQuery := "update foo set time_next = :time_now+if(:min_backoff<<ifnull(epoch, 0) > :max_backoff, :max_backoff, :min_backoff<<ifnull(epoch, 0)), epoch = ifnull(epoch, 0)+1 where id in ::ids and time_acked is null"
-	if query != wantQuery {
-		t.Errorf("GeneratePostponeQuery query: %s, want %s", query, wantQuery)
-	}
-	if _, ok := bv["time_now"]; !ok {
-		t.Errorf("time_now is absent in %v", bv)
-	} else {
-		// time_now cannot be compared.
-		delete(bv, "time_now")
-	}
-	wantbv := map[string]*querypb.BindVariable{
-		"min_backoff": sqltypes.Int64BindVariable(1e9),
-		"max_backoff": sqltypes.Int64BindVariable(4e9),
-		"ids":         wantids,
-	}
-	if !reflect.DeepEqual(bv, wantbv) {
-		t.Errorf("gotid: %v, want %v", bv, wantbv)
-	}
-}
-
 type fakeTabletServer struct {
 	postponeCount sync2.AtomicInt64
 	purgeCount    sync2.AtomicInt64
diff --git a/go/vt/vttablet/tabletserver/schema/engine_test.go b/go/vt/vttablet/tabletserver/schema/engine_test.go
index 62269d7689..2af7a9dc2a 100644
--- a/go/vt/vttablet/tabletserver/schema/engine_test.go
+++ b/go/vt/vttablet/tabletserver/schema/engine_test.go
@@ -402,7 +402,6 @@ func initialSchema() map[string]*Table {
 				}},
 				AckWaitDuration:    30 * time.Second,
 				PurgeAfterDuration: 120 * time.Second,
-				MinBackoff:         30 * time.Second,
 				BatchSize:          1,
 				CacheSize:          10,
 				PollInterval:       30 * time.Second,
diff --git a/go/vt/vttablet/tabletserver/schema/load_table_test.go b/go/vt/vttablet/tabletserver/schema/load_table_test.go
index 0624479e29..02c9af95e1 100644
--- a/go/vt/vttablet/tabletserver/schema/load_table_test.go
+++ b/go/vt/vttablet/tabletserver/schema/load_table_test.go
@@ -123,7 +123,6 @@ func TestLoadTableMessage(t *testing.T) {
 			}},
 			AckWaitDuration:    30 * time.Second,
 			PurgeAfterDuration: 120 * time.Second,
-			MinBackoff:         30 * time.Second,
 			BatchSize:          1,
 			CacheSize:          10,
 			PollInterval:       30 * time.Second,
@@ -131,12 +130,6 @@ func TestLoadTableMessage(t *testing.T) {
 	}
 	assert.Equal(t, want, table)
 
-	// Test loading min/max backoff
-	table, err = newTestLoadTable("USER_TABLE", "vitess_message,vt_ack_wait=30,vt_purge_after=120,vt_batch_size=1,vt_cache_size=10,vt_poller_interval=30,vt_min_backoff=10,vt_max_backoff=100", db)
-	want.MessageInfo.MinBackoff = 10 * time.Second
-	want.MessageInfo.MaxBackoff = 100 * time.Second
-	assert.Equal(t, want, table)
-
 	// Missing property
 	_, err = newTestLoadTable("USER_TABLE", "vitess_message,vt_ack_wait=30", db)
 	wanterr := "not specified for message table"
diff --git a/go/vt/wrangler/traffic_switcher_env_test.go b/go/vt/wrangler/traffic_switcher_env_test.go
index e90f6d8508..9e4dcd63ad 100644
--- a/go/vt/wrangler/traffic_switcher_env_test.go
+++ b/go/vt/wrangler/traffic_switcher_env_test.go
@@ -331,6 +331,8 @@ func (tme *testMigraterEnv) createDBClients(ctx context.Context, t *testing.T) {
 		dbclient := newFakeDBClient()
 		tme.dbSourceClients = append(tme.dbSourceClients, dbclient)
 		dbClientFactory := func() binlogplayer.DBClient { return dbclient }
+		// Replace existing engine with a new one
+		master.Agent.VREngine.Close()
 		master.Agent.VREngine = vreplication.NewEngine(tme.ts, "", master.FakeMysqlDaemon, dbClientFactory, dbclient.DBName())
 		if err := master.Agent.VREngine.Open(ctx); err != nil {
 			t.Fatal(err)
@@ -340,6 +342,8 @@ func (tme *testMigraterEnv) createDBClients(ctx context.Context, t *testing.T) {
 		dbclient := newFakeDBClient()
 		tme.dbTargetClients = append(tme.dbTargetClients, dbclient)
 		dbClientFactory := func() binlogplayer.DBClient { return dbclient }
+		// Replace existing engine with a new one
+		master.Agent.VREngine.Close()
 		master.Agent.VREngine = vreplication.NewEngine(tme.ts, "", master.FakeMysqlDaemon, dbClientFactory, dbclient.DBName())
 		if err := master.Agent.VREngine.Open(ctx); err != nil {
 			t.Fatal(err)
diff --git a/test.go b/test.go
index 284e747078..a31c1ce153 100755
--- a/test.go
+++ b/test.go
@@ -39,6 +39,7 @@ package main
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -85,6 +86,7 @@ var (
 	shard          = flag.Int("shard", -1, "if N>=0, run the tests whose Shard field matches N")
 	tag            = flag.String("tag", "", "if provided, only run tests with the given tag. Can't be combined with -shard or explicit test list")
 	exclude        = flag.String("exclude", "", "if provided, exclude tests containing any of the given tags (comma delimited)")
+	reshard        = flag.Int("rebalance", 0, "if N>0, check the stats and group tests into N similarly-sized bins by average run time")
 	keepData       = flag.Bool("keep-data", false, "don't delete the per-test VTDATAROOT subfolders")
 	printLog       = flag.Bool("print-log", false, "print the log of each failed test (or all tests if -log-pass) to the console")
 	follow         = flag.Bool("follow", false, "print test output as it runs, instead of waiting to see if it passes or fails")
@@ -297,6 +299,22 @@ func main() {
 		log.Fatalf("Can't parse config file: %v", err)
 	}
 
+	// Resharding.
+	if *reshard > 0 {
+		if err := reshardTests(&config, *reshard); err != nil {
+			log.Fatalf("resharding error: %v", err)
+		}
+		log.Printf("Saving updated config...")
+		data, err := json.MarshalIndent(config, "", "\t")
+		if err != nil {
+			log.Fatalf("can't save new config: %v", err)
+		}
+		if err := ioutil.WriteFile(configFileName, data, 0644); err != nil {
+			log.Fatalf("can't write new config: %v", err)
+		}
+		return
+	}
+
 	flavors := []string{"local"}
 
 	if *docker {
@@ -679,6 +697,105 @@ func updateTestStats(name string, update func(*TestStats)) {
 	}
 }
 
+func reshardTests(config *Config, numShards int) error {
+	var stats Stats
+
+	var data []byte
+	if *remoteStats != "" {
+		log.Printf("Using remote stats for resharding: %v", *remoteStats)
+		resp, err := http.Get(*remoteStats)
+		if err != nil {
+			return err
+		}
+		defer resp.Body.Close()
+		if data, err = ioutil.ReadAll(resp.Body); err != nil {
+			return err
+		}
+	} else {
+		var err error
+		data, err = ioutil.ReadFile(statsFileName)
+		if err != nil {
+			return errors.New("can't read stats file")
+		}
+	}
+
+	if err := json.Unmarshal(data, &stats); err != nil {
+		return fmt.Errorf("can't parse stats file: %v", err)
+	}
+
+	// Sort tests by PassTime.
+	var tests []TestStats
+	var totalTime int64
+	for name, test := range stats.TestStats {
+		test.name = name
+		tests = append(tests, test)
+		totalTime += int64(test.PassTime)
+	}
+	sort.Sort(ByPassTime(tests))
+
+	// Group into shards.
+	max := totalTime / int64(numShards)
+	shards := make([][]TestStats, numShards)
+	sums := make([]int64, numShards)
+	// First pass: greedy approximation.
+firstPass:
+	for len(tests) > 0 {
+		v := int64(tests[0].PassTime)
+
+		for n := range shards {
+			if sums[n]+v < max {
+				shards[n] = append(shards[n], tests[0])
+				sums[n] += v
+				tests = tests[1:]
+				continue firstPass
+			}
+		}
+		// None of the bins has room. Go to second pass.
+		break
+	}
+	// Second pass: distribute the remainder.
+	for len(tests) > 0 {
+		nmin := 0
+		min := sums[0]
+
+		for n := range sums {
+			if sums[n] < min {
+				nmin = n
+				min = sums[n]
+			}
+		}
+
+		shards[nmin] = append(shards[nmin], tests[0])
+		sums[nmin] += int64(tests[0].PassTime)
+		tests = tests[1:]
+	}
+
+	// Update config and print results.
+	for i, tests := range shards {
+		for _, t := range tests {
+			ct, ok := config.Tests[t.name]
+			if !ok {
+				log.Printf("WARNING: skipping unknown test: %v", t.name)
+				continue
+			}
+			ct.Shard = i
+			if ct.Args == nil {
+				ct.Args = []string{}
+			}
+			if ct.Command == nil {
+				ct.Command = []string{}
+			}
+			if ct.Tags == nil {
+				ct.Tags = []string{}
+			}
+			log.Printf("%v:\t%v\n", t.name, t.PassTime)
+		}
+		log.Printf("Shard %v total: %v\n", i, time.Duration(sums[i]))
+	}
+
+	return nil
+}
+
 type ByPassTime []TestStats
 
 func (a ByPassTime) Len() int           { return len(a) }