Repository: vitessio/vitess
Pull request #5574 (2019-12-14 23:39:37)

vrepl: support qualified vindex names
This change allows you to use qualified vindex names in the in_keyrange expressions. Example: `in_keyrange(col, 'ks.my_vindex', '-80')`.

The expression can refer to an existing vindex in any keyspace, and we continue to support a straight vindex type as long as it doesn't require additional parameters.


Patch:
diff --git a/.github/workflows/check_make_parser.yml b/.github/workflows/check_make_parser.yml
index efc6c737e7..0657236fc3 100644
--- a/.github/workflows/check_make_parser.yml
+++ b/.github/workflows/check_make_parser.yml
@@ -3,7 +3,7 @@ on: [push, pull_request]
 jobs:
 
   build:
-    name: Check Make Parser
+    name: Build
     runs-on: ubuntu-latest
     steps:
 
@@ -17,7 +17,6 @@ jobs:
 
     - name: Get dependencies
       run: |
-        sudo apt-get update || echo "update failed" 
         sudo apt-get install -y mysql-server mysql-client make unzip g++ etcd curl git wget
         sudo service mysql stop
         sudo service etcd stop
diff --git a/.github/workflows/cluster_endtoend.yml b/.github/workflows/cluster_endtoend.yml
index 5d31167f70..a2f3f9d7d3 100644
--- a/.github/workflows/cluster_endtoend.yml
+++ b/.github/workflows/cluster_endtoend.yml
@@ -3,7 +3,7 @@ on: [push, pull_request]
 jobs:
 
   build:
-    name: Cluster End-to-End Test
+    name: Build
     runs-on: ubuntu-latest
     steps:
 
@@ -17,7 +17,6 @@ jobs:
 
     - name: Get dependencies
       run: |
-        sudo apt-get update || echo "update failed"
         sudo apt-get install -y mysql-server mysql-client make unzip g++ etcd curl git wget
         sudo service mysql stop
         sudo service etcd stop
diff --git a/.github/workflows/endtoend.yml b/.github/workflows/endtoend.yml
index b10977d486..6ebec31c3b 100644
--- a/.github/workflows/endtoend.yml
+++ b/.github/workflows/endtoend.yml
@@ -3,7 +3,7 @@ on: [push, pull_request]
 jobs:
 
   build:
-    name: End-to-End Test
+    name: Build
     runs-on: ubuntu-latest
     steps:
 
@@ -17,7 +17,6 @@ jobs:
 
     - name: Get dependencies
       run: |
-        sudo apt-get update || echo "update failed"
         sudo apt-get install -y mysql-server mysql-client make unzip g++ etcd curl git wget
         sudo service mysql stop
         sudo service etcd stop
diff --git a/.github/workflows/local_example.yml b/.github/workflows/local_example.yml
index b08ce73f88..fe0035ec10 100644
--- a/.github/workflows/local_example.yml
+++ b/.github/workflows/local_example.yml
@@ -3,7 +3,7 @@ on: [push, pull_request]
 jobs:
 
   build:
-    name: Local Example Test
+    name: Build
     runs-on: ubuntu-latest
     steps:
 
@@ -17,7 +17,6 @@ jobs:
 
     - name: Get dependencies
       run: |
-        sudo apt-get update || echo "update failed"
         sudo apt-get install -y mysql-server mysql-client make unzip g++ etcd curl git wget
         sudo service mysql stop
         sudo service etcd stop
diff --git a/.github/workflows/unit.yml b/.github/workflows/unit.yml
index c60e862473..dd53b59a1d 100644
--- a/.github/workflows/unit.yml
+++ b/.github/workflows/unit.yml
@@ -3,7 +3,7 @@ on: [push, pull_request]
 jobs:
 
   build:
-    name: Unit Test
+    name: Build
     runs-on: ubuntu-latest
     steps:
 
@@ -17,14 +17,11 @@ jobs:
 
     - name: Get dependencies
       run: |
-        sudo apt-get update || echo "update failed"
-        sudo apt-get install -y mysql-server mysql-client make unzip g++ curl git wget ant openjdk-8-jdk
+        sudo apt-get install -y mysql-server mysql-client make unzip g++ etcd curl git wget ant openjdk-8-jdk
         sudo service mysql stop
+        sudo service etcd stop
         sudo ln -s /etc/apparmor.d/usr.sbin.mysqld /etc/apparmor.d/disable/
         sudo apparmor_parser -R /etc/apparmor.d/usr.sbin.mysqld
-        mkdir -p dist bin
-        curl -L https://github.com/coreos/etcd/releases/download/v3.3.10/etcd-v3.3.10-linux-amd64.tar.gz | tar -zxC dist
-        mv dist/etcd-v3.3.10-linux-amd64/{etcd,etcdctl} bin/
         go mod download
 
     - name: Run make tools
diff --git a/.github/workflows/unit_race.yml b/.github/workflows/unit_race.yml
index 6f065bf273..a85a6c87ef 100644
--- a/.github/workflows/unit_race.yml
+++ b/.github/workflows/unit_race.yml
@@ -3,7 +3,7 @@ on: [push]
 jobs:
 
   build:
-    name: Unit Test (Race)
+    name: Build
     runs-on: ubuntu-latest
     steps:
 
@@ -17,7 +17,6 @@ jobs:
 
     - name: Get dependencies
       run: |
-        sudo apt-get update || echo "update failed"
         sudo apt-get install -y mysql-server mysql-client make unzip g++ etcd curl git wget
         sudo service mysql stop
         sudo service etcd stop
diff --git a/docker/bootstrap/Dockerfile.mariadb b/docker/bootstrap/Dockerfile.mariadb
index 19d5c9973d..e0f6106837 100644
--- a/docker/bootstrap/Dockerfile.mariadb
+++ b/docker/bootstrap/Dockerfile.mariadb
@@ -22,6 +22,5 @@ RUN for i in $(seq 1 10); do apt-key adv --no-tty --keyserver keys.gnupg.net --r
 # Bootstrap Vitess
 WORKDIR /vt/src/vitess.io/vitess
 
-ENV MYSQL_FLAVOR MariaDB
 USER vitess
 RUN ./bootstrap.sh
diff --git a/docker/bootstrap/Dockerfile.mariadb103 b/docker/bootstrap/Dockerfile.mariadb103
index c2828ddd25..024fe6a80b 100644
--- a/docker/bootstrap/Dockerfile.mariadb103
+++ b/docker/bootstrap/Dockerfile.mariadb103
@@ -11,6 +11,5 @@ RUN apt-key adv --no-tty --recv-keys --keyserver keyserver.ubuntu.com 0xF1656F24
 # Bootstrap Vitess
 WORKDIR /vt/src/vitess.io/vitess
 
-ENV MYSQL_FLAVOR MariaDB103
 USER vitess
 RUN ./bootstrap.sh
diff --git a/docker/bootstrap/Dockerfile.mysql80 b/docker/bootstrap/Dockerfile.mysql80
index 0523a7847c..e6d00bc524 100644
--- a/docker/bootstrap/Dockerfile.mysql80
+++ b/docker/bootstrap/Dockerfile.mysql80
@@ -17,6 +17,5 @@ RUN for i in $(seq 1 10); do apt-key adv --no-tty --recv-keys --keyserver ha.poo
 # Bootstrap Vitess
 WORKDIR /vt/src/vitess.io/vitess
 
-ENV MYSQL_FLAVOR MySQL80
 USER vitess
-RUN ./bootstrap.sh
+RUN ./bootstrap.sh
\ No newline at end of file
diff --git a/docker/bootstrap/Dockerfile.percona80 b/docker/bootstrap/Dockerfile.percona80
index 1ce9c52103..c5ce5d5ee4 100644
--- a/docker/bootstrap/Dockerfile.percona80
+++ b/docker/bootstrap/Dockerfile.percona80
@@ -32,6 +32,5 @@ RUN for i in $(seq 1 10); do apt-key adv --no-tty --keyserver keys.gnupg.net --r
 # Bootstrap Vitess
 WORKDIR /vt/src/vitess.io/vitess
 
-ENV MYSQL_FLAVOR MySQL80
 USER vitess
 RUN ./bootstrap.sh
diff --git a/docker/lite/Dockerfile.alpine b/docker/lite/Dockerfile.alpine
index f57b9d7a8f..b8fc9467f9 100644
--- a/docker/lite/Dockerfile.alpine
+++ b/docker/lite/Dockerfile.alpine
@@ -25,7 +25,6 @@ RUN echo '@edge http://nl.alpinelinux.org/alpine/edge/main' >> /etc/apk/reposito
 ENV VTROOT /vt/src/vitess.io/vitess
 ENV VTDATAROOT /vt/vtdataroot
 ENV PATH $VTROOT/bin:$PATH
-ENV MYSQL_FLAVOR MariaDB103
 
 # Create vitess user
 RUN addgroup -S vitess && adduser -S -G vitess vitess && mkdir -p /vt
diff --git a/docker/lite/Dockerfile.mariadb b/docker/lite/Dockerfile.mariadb
index 6907534c03..77eddfd800 100644
--- a/docker/lite/Dockerfile.mariadb
+++ b/docker/lite/Dockerfile.mariadb
@@ -36,7 +36,6 @@ RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-ins
 ENV VTROOT /vt/src/vitess.io/vitess
 ENV VTDATAROOT /vt/vtdataroot
 ENV PATH $VTROOT/bin:$PATH
-ENV MYSQL_FLAVOR MariaDB
 
 # Copy binaries (placed by build.sh)
 COPY --from=staging /vt/ /vt/
diff --git a/docker/lite/Dockerfile.mariadb103 b/docker/lite/Dockerfile.mariadb103
index cd962803f7..a1d37b4c42 100644
--- a/docker/lite/Dockerfile.mariadb103
+++ b/docker/lite/Dockerfile.mariadb103
@@ -35,7 +35,6 @@ RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-ins
 ENV VTROOT /vt/src/vitess.io/vitess
 ENV VTDATAROOT /vt/vtdataroot
 ENV PATH $VTROOT/bin:$PATH
-ENV MYSQL_FLAVOR MariaDB103
 
 # Copy binaries (placed by build.sh)
 COPY --from=staging /vt/ /vt/
diff --git a/docker/lite/Dockerfile.mysql80 b/docker/lite/Dockerfile.mysql80
index 85383be648..0fade69bd3 100644
--- a/docker/lite/Dockerfile.mysql80
+++ b/docker/lite/Dockerfile.mysql80
@@ -39,7 +39,6 @@ RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-ins
 ENV VTROOT /vt/src/vitess.io/vitess
 ENV VTDATAROOT /vt/vtdataroot
 ENV PATH $VTROOT/bin:$PATH
-ENV MYSQL_FLAVOR MySQL80
 
 # Copy binaries (placed by build.sh)
 COPY --from=staging /vt/ /vt/
diff --git a/docker/lite/Dockerfile.percona80 b/docker/lite/Dockerfile.percona80
index 98b4b30a57..5d872499cf 100644
--- a/docker/lite/Dockerfile.percona80
+++ b/docker/lite/Dockerfile.percona80
@@ -44,7 +44,6 @@ RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-ins
 ENV VTROOT /vt/src/vitess.io/vitess
 ENV VTDATAROOT /vt/vtdataroot
 ENV PATH $VTROOT/bin:$PATH
-ENV MYSQL_FLAVOR MySQL80
 
 # Copy binaries (placed by build.sh)
 COPY --from=staging /vt/ /vt/
diff --git a/go.mod b/go.mod
index af52af7313..44d8c461dd 100644
--- a/go.mod
+++ b/go.mod
@@ -17,7 +17,7 @@ require (
 	github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f // indirect
 	github.com/dchest/safefile v0.0.0-20151022103144-855e8d98f185 // indirect
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible // indirect
-	github.com/evanphx/json-patch v4.5.0+incompatible
+	github.com/evanphx/json-patch v4.5.0+incompatible // indirect
 	github.com/ghodss/yaml v0.0.0-20161207003320-04f313413ffd // indirect
 	github.com/go-ini/ini v1.12.0 // indirect
 	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
@@ -45,7 +45,7 @@ require (
 	github.com/klauspost/crc32 v1.2.0 // indirect
 	github.com/klauspost/pgzip v1.2.0
 	github.com/kr/pretty v0.1.0 // indirect
-	github.com/krishicks/yaml-patch v0.0.10
+	github.com/krishicks/yaml-patch v0.0.10 // indirect
 	github.com/mattn/go-runewidth v0.0.1 // indirect
 	github.com/minio/minio-go v0.0.0-20190131015406-c8a261de75c1
 	github.com/mitchellh/go-testing-interface v1.0.0 // indirect
diff --git a/go/mysql/encoding.go b/go/mysql/encoding.go
index 53b9e2efc7..c9333522cd 100644
--- a/go/mysql/encoding.go
+++ b/go/mysql/encoding.go
@@ -281,17 +281,3 @@ func readLenEncStringAsBytes(data []byte, pos int) ([]byte, int, bool) {
 	}
 	return data[pos : pos+s], pos + s, true
 }
-
-func readLenEncStringAsBytesCopy(data []byte, pos int) ([]byte, int, bool) {
-	size, pos, ok := readLenEncInt(data, pos)
-	if !ok {
-		return nil, 0, false
-	}
-	s := int(size)
-	if pos+s-1 >= len(data) {
-		return nil, 0, false
-	}
-	result := make([]byte, size)
-	copy(result, data[pos:pos+s])
-	return result, pos + s, true
-}
diff --git a/go/mysql/query.go b/go/mysql/query.go
index f6435bf4b6..6ff9d1e9bc 100644
--- a/go/mysql/query.go
+++ b/go/mysql/query.go
@@ -261,7 +261,7 @@ func (c *Conn) parseRow(data []byte, fields []*querypb.Field) ([]sqltypes.Value,
 		}
 		var s []byte
 		var ok bool
-		s, pos, ok = readLenEncStringAsBytesCopy(data, pos)
+		s, pos, ok = readLenEncStringAsBytes(data, pos)
 		if !ok {
 			return nil, NewSQLError(CRMalformedPacket, SSUnknownSQLState, "decoding string failed")
 		}
@@ -823,7 +823,7 @@ func (c *Conn) parseStmtArgs(data []byte, typ querypb.Type, pos int) (sqltypes.V
 		}
 	case sqltypes.Decimal, sqltypes.Text, sqltypes.Blob, sqltypes.VarChar, sqltypes.VarBinary, sqltypes.Char,
 		sqltypes.Bit, sqltypes.Enum, sqltypes.Set, sqltypes.Geometry, sqltypes.Binary, sqltypes.TypeJSON:
-		val, pos, ok := readLenEncStringAsBytesCopy(data, pos)
+		val, pos, ok := readLenEncStringAsBytes(data, pos)
 		return sqltypes.MakeTrusted(sqltypes.VarBinary, val), pos, ok
 	default:
 		return sqltypes.NULL, pos, false
diff --git a/go/vt/sqlparser/comments.go b/go/vt/sqlparser/comments.go
index afffd326c8..26c664cf91 100644
--- a/go/vt/sqlparser/comments.go
+++ b/go/vt/sqlparser/comments.go
@@ -123,9 +123,7 @@ func SplitMarginComments(sql string) (query string, comments MarginComments) {
 		Leading:  strings.TrimLeftFunc(sql[:leadingEnd], unicode.IsSpace),
 		Trailing: strings.TrimRightFunc(sql[trailingStart:], unicode.IsSpace),
 	}
-	return strings.TrimFunc(sql[leadingEnd:trailingStart], func(c rune) bool {
-		return unicode.IsSpace(c) || c == ';'
-	}), comments
+	return strings.TrimFunc(sql[leadingEnd:trailingStart], unicode.IsSpace), comments
 }
 
 // StripLeadingComments trims the SQL string and removes any leading comments
diff --git a/go/vt/topo/etcd2topo/server.go b/go/vt/topo/etcd2topo/server.go
index 5d67a6f010..8e6b2588a4 100644
--- a/go/vt/topo/etcd2topo/server.go
+++ b/go/vt/topo/etcd2topo/server.go
@@ -34,21 +34,13 @@ We follow these conventions within this package:
 package etcd2topo
 
 import (
-	"flag"
 	"strings"
 	"time"
 
 	"github.com/coreos/etcd/clientv3"
-	"github.com/coreos/etcd/pkg/transport"
 	"vitess.io/vitess/go/vt/topo"
 )
 
-var (
-	clientCertPath = flag.String("topo_etcd_tls_cert", "", "path to the client cert to use to connect to the etcd topo server, requires topo_etcd_tls_key, enables TLS")
-	clientKeyPath  = flag.String("topo_etcd_tls_key", "", "path to the client key to use to connect to the etcd topo server, enables TLS")
-	serverCaPath   = flag.String("topo_etcd_tls_ca", "", "path to the ca to use to validate the server cert when connecting to the etcd topo server")
-)
-
 // Factory is the consul topo.Factory implementation.
 type Factory struct{}
 
@@ -79,31 +71,12 @@ func (s *Server) Close() {
 	s.cli = nil
 }
 
-func NewServerWithOpts(serverAddr, root, certPath, keyPath, caPath string) (*Server, error) {
-	// TODO: Rename this to NewServer and change NewServer to a name that signifies it uses the process-wide TLS settings.
-
-	config := clientv3.Config{
+// NewServer returns a new etcdtopo.Server.
+func NewServer(serverAddr, root string) (*Server, error) {
+	cli, err := clientv3.New(clientv3.Config{
 		Endpoints:   strings.Split(serverAddr, ","),
 		DialTimeout: 5 * time.Second,
-	}
-
-	// If TLS is enabled, attach TLS config info.
-	if certPath != "" && keyPath != "" {
-		// Safe now to build up TLS info.
-		tlsInfo := transport.TLSInfo{
-			CertFile:      certPath,
-			KeyFile:       keyPath,
-			TrustedCAFile: caPath,
-		}
-
-		tlsConfig, err := tlsInfo.ClientConfig()
-		if err != nil {
-			return nil, err
-		}
-		config.TLS = tlsConfig
-	}
-
-	cli, err := clientv3.New(config)
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -114,13 +87,6 @@ func NewServerWithOpts(serverAddr, root, certPath, keyPath, caPath string) (*Ser
 	}, nil
 }
 
-// NewServer returns a new etcdtopo.Server.
-func NewServer(serverAddr, root string) (*Server, error) {
-	// TODO: Rename this to a name to signifies this function uses the process-wide TLS settings.
-
-	return NewServerWithOpts(serverAddr, root, *clientCertPath, *clientKeyPath, *serverCaPath)
-}
-
 func init() {
 	topo.RegisterFactory("etcd2", Factory{})
 }
diff --git a/go/vt/vitessdriver/doc.go b/go/vt/vitessdriver/doc.go
index 31ea98651b..a3e2770b8a 100644
--- a/go/vt/vitessdriver/doc.go
+++ b/go/vt/vitessdriver/doc.go
@@ -28,12 +28,13 @@ Example
 Using this SQL driver is as simple as:
 
   import (
+    "time"
     "vitess.io/vitess/go/vt/vitessdriver"
   )
 
   func main() {
     // Connect to vtgate.
-    db, err := vitessdriver.Open("localhost:15991", "@master")
+    db, err := vitessdriver.Open("localhost:15991", "keyspace", "master", 30*time.Second)
 
     // Use "db" via the Golang sql interface.
   }
diff --git a/go/vt/vtgate/executor.go b/go/vt/vtgate/executor.go
index 63e3a0bdfb..7a91df5e65 100644
--- a/go/vt/vtgate/executor.go
+++ b/go/vt/vtgate/executor.go
@@ -845,14 +845,6 @@ func (e *Executor) handleShow(ctx context.Context, safeSession *SafeSession, sql
 				}
 			}
 		}
-	case sqlparser.KeywordString(sqlparser.COLUMNS):
-		if !show.OnTable.Qualifier.IsEmpty() {
-			destKeyspace = show.OnTable.Qualifier.String()
-			show.OnTable.Qualifier = sqlparser.NewTableIdent("")
-		} else {
-			break
-		}
-		sql = sqlparser.String(show)
 	case sqlparser.KeywordString(sqlparser.TABLES):
 		if show.ShowTablesOpt != nil && show.ShowTablesOpt.DbName != "" {
 			if destKeyspace == "" {
diff --git a/go/vt/vtgate/planbuilder/route_option.go b/go/vt/vtgate/planbuilder/route_option.go
index fa3bab2c6a..1d6ba1c367 100644
--- a/go/vt/vtgate/planbuilder/route_option.go
+++ b/go/vt/vtgate/planbuilder/route_option.go
@@ -87,8 +87,12 @@ func (ro *routeOption) JoinCanMerge(pb *primitiveBuilder, rro *routeOption, ajoi
 }
 
 func (ro *routeOption) MergeJoin(rro *routeOption, isLeftJoin bool) {
-	ro.merge(rro)
 	ro.vschemaTable = nil
+	ro.substitutions = append(ro.substitutions, rro.substitutions...)
+	if isLeftJoin {
+		return
+	}
+	// Add RHS vindexes only if it's not a left join.
 	for c, v := range rro.vindexMap {
 		if ro.vindexMap == nil {
 			ro.vindexMap = make(map[*column]vindexes.SingleColumn)
@@ -97,16 +101,6 @@ func (ro *routeOption) MergeJoin(rro *routeOption, isLeftJoin bool) {
 	}
 }
 
-// merge merges two routeOptions. If the LHS (ro) is a SelectReference,
-// then the RHS option values supersede LHS.
-func (ro *routeOption) merge(rro *routeOption) {
-	if ro.eroute.Opcode == engine.SelectReference {
-		// Swap the values and then merge.
-		*ro, *rro = *rro, *ro
-	}
-	ro.substitutions = append(ro.substitutions, rro.substitutions...)
-}
-
 func (ro *routeOption) SubqueryCanMerge(pb *primitiveBuilder, inner *routeOption) bool {
 	return ro.canMerge(inner, func() bool {
 		switch vals := inner.condition.(type) {
@@ -120,7 +114,7 @@ func (ro *routeOption) SubqueryCanMerge(pb *primitiveBuilder, inner *routeOption
 }
 
 func (ro *routeOption) MergeSubquery(subqueryOption *routeOption) {
-	ro.merge(subqueryOption)
+	ro.substitutions = append(ro.substitutions, subqueryOption.substitutions...)
 }
 
 func (ro *routeOption) UnionCanMerge(rro *routeOption) bool {
@@ -128,8 +122,8 @@ func (ro *routeOption) UnionCanMerge(rro *routeOption) bool {
 }
 
 func (ro *routeOption) MergeUnion(rro *routeOption) {
-	ro.merge(rro)
 	ro.vschemaTable = nil
+	ro.substitutions = append(ro.substitutions, rro.substitutions...)
 }
 
 func (ro *routeOption) SubqueryToTable(rb *route, vindexMap map[*column]vindexes.SingleColumn) {
@@ -155,7 +149,9 @@ func (ro *routeOption) canMerge(rro *routeOption, customCheck func() bool) bool
 			return true
 		}
 	case engine.SelectReference:
-		return true
+		// TODO(sougou): this can be changed to true, but we'll have
+		// to merge against rro insteal of ro.
+		return false
 	case engine.SelectNext:
 		return false
 	}
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vcopier_flaky_test.go b/go/vt/vttablet/tabletmanager/vreplication/vcopier_test.go
similarity index 100%
rename from go/vt/vttablet/tabletmanager/vreplication/vcopier_flaky_test.go
rename to go/vt/vttablet/tabletmanager/vreplication/vcopier_test.go
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vstreamer_client.go b/go/vt/vttablet/tabletmanager/vreplication/vstreamer_client.go
index 3f30323820..8e4f7a3b84 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vstreamer_client.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vstreamer_client.go
@@ -27,7 +27,6 @@ import (
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/dbconfigs"
 	"vitess.io/vitess/go/vt/grpcclient"
-	"vitess.io/vitess/go/vt/vtgate/vindexes"
 	"vitess.io/vitess/go/vt/vttablet/queryservice"
 	"vitess.io/vitess/go/vt/vttablet/tabletconn"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/connpool"
@@ -187,7 +186,7 @@ func (vsClient *MySQLVStreamerClient) VStream(ctx context.Context, startPos stri
 	if !vsClient.isOpen {
 		return errors.New("can't VStream without opening client")
 	}
-	streamer := vstreamer.NewVStreamer(ctx, vsClient.sourceConnParams, vsClient.sourceSe, startPos, filter, &vindexes.KeyspaceSchema{}, send)
+	streamer := vstreamer.NewVStreamer(ctx, vsClient.sourceConnParams, vsClient.sourceSe, startPos, filter, nil, send)
 	return streamer.Stream()
 }
 
@@ -204,7 +203,7 @@ func (vsClient *MySQLVStreamerClient) VStreamRows(ctx context.Context, query str
 		}
 		row = r.Rows[0]
 	}
-	streamer := vstreamer.NewRowStreamer(ctx, vsClient.sourceConnParams, vsClient.sourceSe, query, row, &vindexes.KeyspaceSchema{}, send)
+	streamer := vstreamer.NewRowStreamer(ctx, vsClient.sourceConnParams, vsClient.sourceSe, query, row, nil, send)
 	return streamer.Stream()
 }
 
diff --git a/go/vt/vttablet/tabletserver/vstreamer/engine.go b/go/vt/vttablet/tabletserver/vstreamer/engine.go
index 19b54c572d..2b2819db61 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/engine.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/engine.go
@@ -49,7 +49,7 @@ type Engine struct {
 	// cp is initialized by InitDBConfig
 	cp *mysql.ConnParams
 
-	// mu protects isOpen, streamers, streamIdx and kschema.
+	// mu protects isOpen, streamers, streamIdx and vschema.
 	mu sync.Mutex
 
 	isOpen bool
@@ -61,12 +61,12 @@ type Engine struct {
 	resultStreamers map[int]*resultStreamer
 	streamIdx       int
 
-	// watcherOnce is used for initializing kschema
+	// watcherOnce is used for initializing vschema
 	// and setting up the vschema watch. It's guaranteed that
-	// no stream will start until kschema is initialized by
+	// no stream will start until vschema is initialized by
 	// the first call through watcherOnce.
 	watcherOnce sync.Once
-	kschema     *vindexes.KeyspaceSchema
+	lvschema    *localVSchema
 
 	// The following members are initialized once at the beginning.
 	ts       srvtopo.Server
@@ -83,7 +83,7 @@ func NewEngine(ts srvtopo.Server, se *schema.Engine) *Engine {
 		streamers:       make(map[int]*vstreamer),
 		rowStreamers:    make(map[int]*rowStreamer),
 		resultStreamers: make(map[int]*resultStreamer),
-		kschema:         &vindexes.KeyspaceSchema{},
+		lvschema:        &localVSchema{vschema: &vindexes.VSchema{}},
 		ts:              ts,
 		se:              se,
 	}
@@ -146,15 +146,15 @@ func (vse *Engine) Close() {
 	vse.wg.Wait()
 }
 
-func (vse *Engine) vschema() *vindexes.KeyspaceSchema {
+func (vse *Engine) vschema() *vindexes.VSchema {
 	vse.mu.Lock()
 	defer vse.mu.Unlock()
-	return vse.kschema
+	return vse.lvschema.vschema
 }
 
 // Stream starts a new stream.
 func (vse *Engine) Stream(ctx context.Context, startPos string, filter *binlogdatapb.Filter, send func([]*binlogdatapb.VEvent) error) error {
-	// Ensure kschema is initialized and the watcher is started.
+	// Ensure vschema is initialized and the watcher is started.
 	// Starting of the watcher has to be delayed till the first call to Stream
 	// because this overhead should be incurred only if someone uses this feature.
 	vse.watcherOnce.Do(vse.setWatch)
@@ -166,7 +166,7 @@ func (vse *Engine) Stream(ctx context.Context, startPos string, filter *binlogda
 		if !vse.isOpen {
 			return nil, 0, errors.New("VStreamer is not open")
 		}
-		streamer := NewVStreamer(ctx, vse.cp, vse.se, startPos, filter, vse.kschema, send)
+		streamer := NewVStreamer(ctx, vse.cp, vse.se, startPos, filter, vse.lvschema, send)
 		idx := vse.streamIdx
 		vse.streamers[idx] = streamer
 		vse.streamIdx++
@@ -193,7 +193,7 @@ func (vse *Engine) Stream(ctx context.Context, startPos string, filter *binlogda
 
 // StreamRows streams rows.
 func (vse *Engine) StreamRows(ctx context.Context, query string, lastpk []sqltypes.Value, send func(*binlogdatapb.VStreamRowsResponse) error) error {
-	// Ensure kschema is initialized and the watcher is started.
+	// Ensure vschema is initialized and the watcher is started.
 	// Starting of the watcher has to be delayed till the first call to Stream
 	// because this overhead should be incurred only if someone uses this feature.
 	vse.watcherOnce.Do(vse.setWatch)
@@ -206,7 +206,7 @@ func (vse *Engine) StreamRows(ctx context.Context, query string, lastpk []sqltyp
 		if !vse.isOpen {
 			return nil, 0, errors.New("VStreamer is not open")
 		}
-		rowStreamer := NewRowStreamer(ctx, vse.cp, vse.se, query, lastpk, vse.kschema, send)
+		rowStreamer := NewRowStreamer(ctx, vse.cp, vse.se, query, lastpk, vse.lvschema, send)
 		idx := vse.streamIdx
 		vse.rowStreamers[idx] = rowStreamer
 		vse.streamIdx++
@@ -273,7 +273,7 @@ func (vse *Engine) ServeHTTP(response http.ResponseWriter, request *http.Request
 	}
 	response.Header().Set("Content-Type", "application/json; charset=utf-8")
 	vs := vse.vschema()
-	if vs == nil || vs.Keyspace == nil {
+	if vs == nil {
 		response.Write([]byte("{}"))
 	}
 	b, err := json.MarshalIndent(vs, "", "  ")
@@ -289,39 +289,39 @@ func (vse *Engine) ServeHTTP(response http.ResponseWriter, request *http.Request
 func (vse *Engine) setWatch() {
 	// WatchSrvVSchema does not return until the inner func has been called at least once.
 	vse.ts.WatchSrvVSchema(context.TODO(), vse.cell, func(v *vschemapb.SrvVSchema, err error) {
-		var kschema *vindexes.KeyspaceSchema
 		switch {
 		case err == nil:
-			kschema, err = vindexes.BuildKeyspaceSchema(v.Keyspaces[vse.keyspace], vse.keyspace)
-			if err != nil {
-				log.Errorf("Error building vschema %s: %v", vse.keyspace, err)
-				vschemaErrors.Add(1)
-				return
-			}
+			// Build vschema down below.
 		case topo.IsErrType(err, topo.NoNode):
-			// No-op.
+			v = nil
 		default:
-			log.Errorf("Error fetching vschema %s: %v", vse.keyspace, err)
+			log.Errorf("Error fetching vschema: %v", err)
 			vschemaErrors.Add(1)
 			return
 		}
-
-		if kschema == nil {
-			kschema = &vindexes.KeyspaceSchema{
-				Keyspace: &vindexes.Keyspace{
-					Name: vse.keyspace,
-				},
+		var vschema *vindexes.VSchema
+		if v != nil {
+			vschema, err = vindexes.BuildVSchema(v)
+			if err != nil {
+				log.Errorf("Error building vschema: %v", err)
+				vschemaErrors.Add(1)
+				return
 			}
+		} else {
+			vschema = &vindexes.VSchema{}
 		}
 
 		// Broadcast the change to all streamers.
 		vse.mu.Lock()
 		defer vse.mu.Unlock()
-		vse.kschema = kschema
-		b, _ := json.MarshalIndent(kschema, "", "  ")
-		log.Infof("Updated KSchema: %s", b)
+		vse.lvschema = &localVSchema{
+			keyspace: vse.keyspace,
+			vschema:  vschema,
+		}
+		b, _ := json.MarshalIndent(vschema, "", "  ")
+		log.Infof("Updated vschema: %s", b)
 		for _, s := range vse.streamers {
-			s.SetKSchema(kschema)
+			s.SetVSchema(vse.lvschema)
 		}
 		vschemaUpdates.Add(1)
 	})
diff --git a/go/vt/vttablet/tabletserver/vstreamer/local_vschema.go b/go/vt/vttablet/tabletserver/vstreamer/local_vschema.go
new file mode 100644
index 0000000000..e6e8c61eaa
--- /dev/null
+++ b/go/vt/vttablet/tabletserver/vstreamer/local_vschema.go
@@ -0,0 +1,68 @@
+/*
+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 vstreamer
+
+import (
+	"fmt"
+	"strings"
+
+	"vitess.io/vitess/go/vt/vtgate/vindexes"
+)
+
+// localVSchema provides vschema behavior specific to vstreamer.
+// Tables are searched within keyspace, but vindexes can be referenced
+// outside the current keyspace.
+type localVSchema struct {
+	keyspace string
+	vschema  *vindexes.VSchema
+}
+
+func (lvs *localVSchema) FindTable(tablename string) (*vindexes.Table, error) {
+	ks, ok := lvs.vschema.Keyspaces[lvs.keyspace]
+	if !ok {
+		return nil, fmt.Errorf("keyspace %s not found in vschema", lvs.keyspace)
+	}
+	table := ks.Tables[tablename]
+	if table == nil {
+		return nil, fmt.Errorf("table %s not found", tablename)
+	}
+	return table, nil
+}
+
+func (lvs *localVSchema) FindOrCreateVindex(qualifiedName string) (vindexes.Vindex, error) {
+	splits := strings.Split(qualifiedName, ".")
+	var keyspace, name string
+	switch len(splits) {
+	case 1:
+		name = splits[0]
+	case 2:
+		keyspace, name = splits[0], splits[1]
+	default:
+		return nil, fmt.Errorf("invalid vindex name: %v", qualifiedName)
+	}
+	vindex, err := lvs.vschema.FindVindex(keyspace, name)
+	if err != nil {
+		return nil, err
+	}
+	if vindex != nil {
+		return vindex, nil
+	}
+	if keyspace != "" {
+		return nil, fmt.Errorf("vindex %v not found", qualifiedName)
+	}
+	return vindexes.CreateVindex(name, name, map[string]string{})
+}
diff --git a/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go b/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
index 8e8f211cde..dd5a58be79 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
@@ -194,7 +194,7 @@ func tableMatches(table sqlparser.TableName, dbname string, filter *binlogdatapb
 	return false
 }
 
-func buildPlan(ti *Table, kschema *vindexes.KeyspaceSchema, filter *binlogdatapb.Filter) (*Plan, error) {
+func buildPlan(ti *Table, vschema *localVSchema, filter *binlogdatapb.Filter) (*Plan, error) {
 	for _, rule := range filter.Rules {
 		switch {
 		case strings.HasPrefix(rule.Match, "/"):
@@ -206,15 +206,15 @@ func buildPlan(ti *Table, kschema *vindexes.KeyspaceSchema, filter *binlogdatapb
 			if !result {
 				continue
 			}
-			return buildREPlan(ti, kschema, rule.Filter)
+			return buildREPlan(ti, vschema, rule.Filter)
 		case rule.Match == ti.Name:
-			return buildTablePlan(ti, kschema, rule.Filter)
+			return buildTablePlan(ti, vschema, rule.Filter)
 		}
 	}
 	return nil, nil
 }
 
-func buildREPlan(ti *Table, kschema *vindexes.KeyspaceSchema, filter string) (*Plan, error) {
+func buildREPlan(ti *Table, vschema *localVSchema, filter string) (*Plan, error) {
 	plan := &Plan{
 		Table: ti,
 	}
@@ -230,17 +230,15 @@ func buildREPlan(ti *Table, kschema *vindexes.KeyspaceSchema, filter string) (*P
 
 	// We need to additionally set VindexColumn, Vindex and KeyRange
 	// based on the Primary Vindex of the table.
-	// Find table in kschema.
-	table := kschema.Tables[ti.Name]
-	if table == nil {
-		return nil, fmt.Errorf("no vschema definition for table %s", ti.Name)
+	table, err := vschema.FindTable(ti.Name)
+	if err != nil {
+		return nil, err
 	}
 	// Get Primary Vindex.
 	if len(table.ColumnVindexes) == 0 {
 		return nil, fmt.Errorf("table %s has no primary vindex", ti.Name)
 	}
 	plan.Vindex = table.ColumnVindexes[0].Vindex
-	var err error
 	plan.VindexColumns, err = buildVindexColumns(plan.Table, table.ColumnVindexes[0].Columns)
 	if err != nil {
 		return nil, err
@@ -258,7 +256,7 @@ func buildREPlan(ti *Table, kschema *vindexes.KeyspaceSchema, filter string) (*P
 	return plan, nil
 }
 
-func buildTablePlan(ti *Table, kschema *vindexes.KeyspaceSchema, query string) (*Plan, error) {
+func buildTablePlan(ti *Table, vschema *localVSchema, query string) (*Plan, error) {
 	sel, fromTable, err := analyzeSelect(query)
 	if err != nil {
 		return nil, err
@@ -270,7 +268,7 @@ func buildTablePlan(ti *Table, kschema *vindexes.KeyspaceSchema, query string) (
 	plan := &Plan{
 		Table: ti,
 	}
-	if err := plan.analyzeExprs(kschema, sel.SelectExprs); err != nil {
+	if err := plan.analyzeExprs(vschema, sel.SelectExprs); err != nil {
 		return nil, err
 	}
 
@@ -285,7 +283,7 @@ func buildTablePlan(ti *Table, kschema *vindexes.KeyspaceSchema, query string) (
 	if !funcExpr.Name.EqualString("in_keyrange") {
 		return nil, fmt.Errorf("unsupported where clause: %v", sqlparser.String(sel.Where))
 	}
-	if err := plan.analyzeInKeyRange(kschema, funcExpr.Exprs); err != nil {
+	if err := plan.analyzeInKeyRange(vschema, funcExpr.Exprs); err != nil {
 		return nil, err
 	}
 	return plan, nil
@@ -314,10 +312,10 @@ func analyzeSelect(query string) (sel *sqlparser.Select, fromTable sqlparser.Tab
 	return sel, fromTable, nil
 }
 
-func (plan *Plan) analyzeExprs(kschema *vindexes.KeyspaceSchema, selExprs sqlparser.SelectExprs) error {
+func (plan *Plan) analyzeExprs(vschema *localVSchema, selExprs sqlparser.SelectExprs) error {
 	if _, ok := selExprs[0].(*sqlparser.StarExpr); !ok {
 		for _, expr := range selExprs {
-			cExpr, err := plan.analyzeExpr(kschema, expr)
+			cExpr, err := plan.analyzeExpr(vschema, expr)
 			if err != nil {
 				return err
 			}
@@ -337,7 +335,7 @@ func (plan *Plan) analyzeExprs(kschema *vindexes.KeyspaceSchema, selExprs sqlpar
 	return nil
 }
 
-func (plan *Plan) analyzeExpr(kschema *vindexes.KeyspaceSchema, selExpr sqlparser.SelectExpr) (cExpr ColExpr, err error) {
+func (plan *Plan) analyzeExpr(vschema *localVSchema, selExpr sqlparser.SelectExpr) (cExpr ColExpr, err error) {
 	aliased, ok := selExpr.(*sqlparser.AliasedExpr)
 	if !ok {
 		return ColExpr{}, fmt.Errorf("unsupported: %v", sqlparser.String(selExpr))
@@ -367,9 +365,9 @@ func (plan *Plan) analyzeExpr(kschema *vindexes.KeyspaceSchema, selExpr sqlparse
 		if len(inner.Exprs) != 0 {
 			return ColExpr{}, fmt.Errorf("unexpected: %v", sqlparser.String(inner))
 		}
-		table := kschema.Tables[plan.Table.Name]
-		if table == nil {
-			return ColExpr{}, fmt.Errorf("no vschema definition for table %s", plan.Table.Name)
+		table, err := vschema.FindTable(plan.Table.Name)
+		if err != nil {
+			return ColExpr{}, err
 		}
 		// Get Primary Vindex.
 		if len(table.ColumnVindexes) == 0 {
@@ -390,14 +388,14 @@ func (plan *Plan) analyzeExpr(kschema *vindexes.KeyspaceSchema, selExpr sqlparse
 	}
 }
 
-func (plan *Plan) analyzeInKeyRange(kschema *vindexes.KeyspaceSchema, exprs sqlparser.SelectExprs) error {
+func (plan *Plan) analyzeInKeyRange(vschema *localVSchema, exprs sqlparser.SelectExprs) error {
 	var colnames []sqlparser.ColIdent
 	var krExpr sqlparser.SelectExpr
 	switch {
 	case len(exprs) == 1:
-		table := kschema.Tables[plan.Table.Name]
-		if table == nil {
-			return fmt.Errorf("no vschema definition for table %s", plan.Table.Name)
+		table, err := vschema.FindTable(plan.Table.Name)
+		if err != nil {
+			return err
 		}
 		// Get Primary Vindex.
 		if len(table.ColumnVindexes) == 0 {
@@ -426,7 +424,7 @@ func (plan *Plan) analyzeInKeyRange(kschema *vindexes.KeyspaceSchema, exprs sqlp
 		if err != nil {
 			return err
 		}
-		plan.Vindex, err = vindexes.CreateVindex(vtype, vtype, map[string]string{})
+		plan.Vindex, err = vschema.FindOrCreateVindex(vtype)
 		if err != nil {
 			return err
 		}
diff --git a/go/vt/vttablet/tabletserver/vstreamer/rowstreamer.go b/go/vt/vttablet/tabletserver/vstreamer/rowstreamer.go
index 3bdf0f2bda..6ad3a9b6b2 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/rowstreamer.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/rowstreamer.go
@@ -25,7 +25,6 @@ import (
 	"vitess.io/vitess/go/vt/dbconfigs"
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/sqlparser"
-	"vitess.io/vitess/go/vt/vtgate/vindexes"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/schema"
 
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
@@ -41,14 +40,14 @@ type rowStreamer struct {
 	query   string
 	lastpk  []sqltypes.Value
 	send    func(*binlogdatapb.VStreamRowsResponse) error
-	kschema *vindexes.KeyspaceSchema
+	vschema *localVSchema
 
 	plan      *Plan
 	pkColumns []int
 	sendQuery string
 }
 
-func NewRowStreamer(ctx context.Context, cp *mysql.ConnParams, se *schema.Engine, query string, lastpk []sqltypes.Value, kschema *vindexes.KeyspaceSchema, send func(*binlogdatapb.VStreamRowsResponse) error) *rowStreamer {
+func NewRowStreamer(ctx context.Context, cp *mysql.ConnParams, se *schema.Engine, query string, lastpk []sqltypes.Value, vschema *localVSchema, send func(*binlogdatapb.VStreamRowsResponse) error) *rowStreamer {
 	ctx, cancel := context.WithCancel(ctx)
 	return &rowStreamer{
 		ctx:     ctx,
@@ -58,7 +57,7 @@ func NewRowStreamer(ctx context.Context, cp *mysql.ConnParams, se *schema.Engine
 		query:   query,
 		lastpk:  lastpk,
 		send:    send,
-		kschema: kschema,
+		vschema: vschema,
 	}
 }
 
@@ -103,7 +102,7 @@ func (rs *rowStreamer) buildPlan() error {
 		Name:    st.Name.String(),
 		Columns: st.Columns,
 	}
-	rs.plan, err = buildTablePlan(ti, rs.kschema, rs.query)
+	rs.plan, err = buildTablePlan(ti, rs.vschema, rs.query)
 	if err != nil {
 		return err
 	}
diff --git a/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go b/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go
index 09c3eb3967..a26f9a4803 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go
@@ -29,7 +29,6 @@ import (
 	"vitess.io/vitess/go/vt/binlog"
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/sqlparser"
-	"vitess.io/vitess/go/vt/vtgate/vindexes"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/schema"
 
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
@@ -53,9 +52,8 @@ type vstreamer struct {
 	filter   *binlogdatapb.Filter
 	send     func([]*binlogdatapb.VEvent) error
 
-	// A kschema is a VSchema for just one keyspace.
-	kevents        chan *vindexes.KeyspaceSchema
-	kschema        *vindexes.KeyspaceSchema
+	vevents        chan *localVSchema
+	vschema        *localVSchema
 	plans          map[uint64]*streamerPlan
 	journalTableID uint64
 
@@ -71,7 +69,7 @@ type streamerPlan struct {
 	TableMap *mysql.TableMap
 }
 
-func NewVStreamer(ctx context.Context, cp *mysql.ConnParams, se *schema.Engine, startPos string, filter *binlogdatapb.Filter, kschema *vindexes.KeyspaceSchema, send func([]*binlogdatapb.VEvent) error) *vstreamer {
+func NewVStreamer(ctx context.Context, cp *mysql.ConnParams, se *schema.Engine, startPos string, filter *binlogdatapb.Filter, vschema *localVSchema, send func([]*binlogdatapb.VEvent) error) *vstreamer {
 	ctx, cancel := context.WithCancel(ctx)
 	return &vstreamer{
 		ctx:      ctx,
@@ -81,18 +79,18 @@ func NewVStreamer(ctx context.Context, cp *mysql.ConnParams, se *schema.Engine,
 		startPos: startPos,
 		filter:   filter,
 		send:     send,
-		kevents:  make(chan *vindexes.KeyspaceSchema, 1),
-		kschema:  kschema,
+		vevents:  make(chan *localVSchema, 1),
+		vschema:  vschema,
 		plans:    make(map[uint64]*streamerPlan),
 	}
 }
 
-// SetKSchema updates all existing against the new kschema.
-func (vs *vstreamer) SetKSchema(kschema *vindexes.KeyspaceSchema) {
+// SetVSchema updates all existing streams against the new vschema.
+func (vs *vstreamer) SetVSchema(vschema *localVSchema) {
 	// Since vs.Stream is a single-threaded loop. We just send an event to
 	// that thread, which helps us avoid mutexes to update the plans.
 	select {
-	case vs.kevents <- kschema:
+	case vs.vevents <- vschema:
 	case <-vs.ctx.Done():
 	}
 }
@@ -223,7 +221,7 @@ func (vs *vstreamer) parseEvents(ctx context.Context, events <-chan mysql.Binlog
 					return fmt.Errorf("error sending event: %v", err)
 				}
 			}
-		case vs.kschema = <-vs.kevents:
+		case vs.vschema = <-vs.vevents:
 			if err := vs.rebuildPlans(); err != nil {
 				return err
 			}
@@ -470,7 +468,7 @@ func (vs *vstreamer) buildTablePlan(id uint64, tm *mysql.TableMap) (*binlogdatap
 		Name:    tm.Name,
 		Columns: cols,
 	}
-	plan, err := buildPlan(table, vs.kschema, vs.filter)
+	plan, err := buildPlan(table, vs.vschema, vs.filter)
 	if err != nil {
 		return nil, err
 	}
@@ -600,11 +598,11 @@ func (vs *vstreamer) processRowEvent(vevents []*binlogdatapb.VEvent, plan *strea
 func (vs *vstreamer) rebuildPlans() error {
 	for id, plan := range vs.plans {
 		if plan == nil {
-			// If a table has no plan, a kschema change will not
+			// If a table has no plan, a vschema change will not
 			// cause that to change.
 			continue
 		}
-		newPlan, err := buildPlan(plan.Table, vs.kschema, vs.filter)
+		newPlan, err := buildPlan(plan.Table, vs.vschema, vs.filter)
 		if err != nil {
 			return err
 		}


Test patch:
diff --git a/.github/workflows/e2e_race.yml b/.github/workflows/e2e_race.yml
index 2e6fb4f493..0439d06711 100644
--- a/.github/workflows/e2e_race.yml
+++ b/.github/workflows/e2e_race.yml
@@ -3,7 +3,7 @@ on: [push, pull_request]
 jobs:
 
   build:
-    name: End-to-End Test (Race)
+    name: Build
     runs-on: ubuntu-latest
     steps:
 
@@ -17,7 +17,6 @@ jobs:
 
     - name: Get dependencies
       run: |
-        sudo apt-get update || echo "update failed"
         sudo apt-get install -y mysql-server mysql-client make unzip g++ etcd curl git wget
         sudo service mysql stop
         sudo service etcd stop
diff --git a/go/mysql/encoding_test.go b/go/mysql/encoding_test.go
index ac38b5c7db..2ca9a84c93 100644
--- a/go/mysql/encoding_test.go
+++ b/go/mysql/encoding_test.go
@@ -303,24 +303,6 @@ func TestEncString(t *testing.T) {
 			t.Errorf("readLenEncStringAsBytes returned ok=true for empty value %v", test.value)
 		}
 
-		// Check successful decoding as bytes.
-		gotbcopy, posCopy, ok := readLenEncStringAsBytesCopy(test.lenEncoded, 0)
-		if !ok || string(gotb) != test.value || pos != len(test.lenEncoded) {
-			t.Errorf("readLenEncString returned %v/%v/%v but expected %v/%v/%v", gotbcopy, posCopy, ok, test.value, len(test.lenEncoded), true)
-		}
-
-		// Check failed decoding as bytes with shorter data.
-		_, _, ok = readLenEncStringAsBytesCopy(test.lenEncoded[:len(test.lenEncoded)-1], 0)
-		if ok {
-			t.Errorf("readLenEncStringAsBytes returned ok=true for shorter value %v", test.value)
-		}
-
-		// Check failed decoding as bytes with no data.
-		_, _, ok = readLenEncStringAsBytesCopy([]byte{}, 0)
-		if ok {
-			t.Errorf("readLenEncStringAsBytes returned ok=true for empty value %v", test.value)
-		}
-
 		// null encoded tests.
 
 		// Check successful encoding.
diff --git a/go/vt/discovery/healthcheck_flaky_test.go b/go/vt/discovery/healthcheck_test.go
similarity index 100%
rename from go/vt/discovery/healthcheck_flaky_test.go
rename to go/vt/discovery/healthcheck_test.go
diff --git a/go/vt/logutil/logutil_flaky_test.go b/go/vt/logutil/logutil_test.go
similarity index 100%
rename from go/vt/logutil/logutil_flaky_test.go
rename to go/vt/logutil/logutil_test.go
diff --git a/go/vt/sqlparser/analyzer_test.go b/go/vt/sqlparser/analyzer_test.go
index f97f36ccbc..8ecc3ce0db 100644
--- a/go/vt/sqlparser/analyzer_test.go
+++ b/go/vt/sqlparser/analyzer_test.go
@@ -50,9 +50,6 @@ func TestPreview(t *testing.T) {
 		{"begin ...", StmtUnknown},
 		{"begin /* ... */", StmtBegin},
 		{"begin /* ... *//*test*/", StmtBegin},
-		{"begin;", StmtBegin},
-		{"begin ;", StmtBegin},
-		{"begin; /*...*/", StmtBegin},
 		{"start transaction", StmtBegin},
 		{"commit", StmtCommit},
 		{"commit /*...*/", StmtCommit},
diff --git a/go/vt/sqlparser/parse_test.go b/go/vt/sqlparser/parse_test.go
index 8ae27a8594..564edccefc 100644
--- a/go/vt/sqlparser/parse_test.go
+++ b/go/vt/sqlparser/parse_test.go
@@ -1442,9 +1442,6 @@ var (
 		input: "stream /* comment */ * from t",
 	}, {
 		input: "begin",
-	}, {
-		input:  "begin;",
-		output: "begin",
 	}, {
 		input:  "start transaction",
 		output: "begin",
diff --git a/go/vt/tlstest/tlstest.go b/go/vt/tlstest/tlstest.go
index 810aa4c279..89b715cf52 100644
--- a/go/vt/tlstest/tlstest.go
+++ b/go/vt/tlstest/tlstest.go
@@ -147,51 +147,3 @@ func CreateSignedCert(root, parent, serial, name, commonName string) {
 		"-extfile", config,
 		"-out", cert)
 }
-
-type ClientServerKeyPairs struct {
-	ServerCert string
-	ServerKey  string
-	ServerCA   string
-	ServerName string
-	ClientCert string
-	ClientKey  string
-	ClientCA   string
-}
-
-var serialCounter = 0
-
-func CreateClientServerCertPairs(root string) ClientServerKeyPairs {
-	// Create the certs and configs.
-	CreateCA(root)
-
-	serverSerial := fmt.Sprintf("%03d", serialCounter*2+1)
-	clientSerial := fmt.Sprintf("%03d", serialCounter*2+2)
-
-	serialCounter = serialCounter + 1
-
-	serverName := fmt.Sprintf("server-%s", serverSerial)
-	serverCACommonName := fmt.Sprintf("Server %s CA", serverSerial)
-	serverCertName := fmt.Sprintf("server-instance-%s", serverSerial)
-	serverCertCommonName := fmt.Sprintf("server%s.example.com", serverSerial)
-
-	clientName := fmt.Sprintf("clients-%s", serverSerial)
-	clientCACommonName := fmt.Sprintf("Clients %s CA", serverSerial)
-	clientCertName := fmt.Sprintf("client-instance-%s", serverSerial)
-	clientCertCommonName := fmt.Sprintf("Client Instance %s", serverSerial)
-
-	CreateSignedCert(root, CA, serverSerial, serverName, serverCACommonName)
-	CreateSignedCert(root, serverName, serverSerial, serverCertName, serverCertCommonName)
-
-	CreateSignedCert(root, CA, clientSerial, clientName, clientCACommonName)
-	CreateSignedCert(root, clientName, serverSerial, clientCertName, clientCertCommonName)
-
-	return ClientServerKeyPairs{
-		ServerCert: path.Join(root, fmt.Sprintf("%s-cert.pem", serverCertName)),
-		ServerKey:  path.Join(root, fmt.Sprintf("%s-key.pem", serverCertName)),
-		ServerCA:   path.Join(root, fmt.Sprintf("%s-cert.pem", serverName)),
-		ClientCert: path.Join(root, fmt.Sprintf("%s-cert.pem", clientCertName)),
-		ClientKey:  path.Join(root, fmt.Sprintf("%s-key.pem", clientCertName)),
-		ClientCA:   path.Join(root, fmt.Sprintf("%s-cert.pem", clientName)),
-		ServerName: serverCertCommonName,
-	}
-}
diff --git a/go/vt/tlstest/tlstest_test.go b/go/vt/tlstest/tlstest_test.go
index 746ab161ce..1f22403cdd 100644
--- a/go/vt/tlstest/tlstest_test.go
+++ b/go/vt/tlstest/tlstest_test.go
@@ -24,6 +24,7 @@ import (
 	"io/ioutil"
 	"net"
 	"os"
+	"path"
 	"strings"
 	"sync"
 	"testing"
@@ -46,20 +47,20 @@ func TestClientServer(t *testing.T) {
 	}
 	defer os.RemoveAll(root)
 
-	clientServerKeyPairs := CreateClientServerCertPairs(root)
+	clientServerKeyPairs := createClientServerCertPairs(root)
 
 	serverConfig, err := vttls.ServerConfig(
-		clientServerKeyPairs.ServerCert,
-		clientServerKeyPairs.ServerKey,
-		clientServerKeyPairs.ClientCA)
+		clientServerKeyPairs.serverCert,
+		clientServerKeyPairs.serverKey,
+		clientServerKeyPairs.clientCA)
 	if err != nil {
 		t.Fatalf("TLSServerConfig failed: %v", err)
 	}
 	clientConfig, err := vttls.ClientConfig(
-		clientServerKeyPairs.ClientCert,
-		clientServerKeyPairs.ClientKey,
-		clientServerKeyPairs.ServerCA,
-		clientServerKeyPairs.ServerName)
+		clientServerKeyPairs.clientCert,
+		clientServerKeyPairs.clientKey,
+		clientServerKeyPairs.serverCA,
+		clientServerKeyPairs.serverName)
 	if err != nil {
 		t.Fatalf("TLSClientConfig failed: %v", err)
 	}
@@ -116,10 +117,10 @@ func TestClientServer(t *testing.T) {
 	//
 
 	badClientConfig, err := vttls.ClientConfig(
-		clientServerKeyPairs.ServerCert,
-		clientServerKeyPairs.ServerKey,
-		clientServerKeyPairs.ServerCA,
-		clientServerKeyPairs.ServerName)
+		clientServerKeyPairs.serverCert,
+		clientServerKeyPairs.serverKey,
+		clientServerKeyPairs.serverCA,
+		clientServerKeyPairs.serverName)
 	if err != nil {
 		t.Fatalf("TLSClientConfig failed: %v", err)
 	}
@@ -164,19 +165,69 @@ func TestClientServer(t *testing.T) {
 	}
 }
 
-func getServerConfig(keypairs ClientServerKeyPairs) (*tls.Config, error) {
+var serialCounter = 0
+
+type clientServerKeyPairs struct {
+	serverCert string
+	serverKey  string
+	serverCA   string
+	serverName string
+	clientCert string
+	clientKey  string
+	clientCA   string
+}
+
+func createClientServerCertPairs(root string) clientServerKeyPairs {
+
+	// Create the certs and configs.
+	CreateCA(root)
+
+	serverSerial := fmt.Sprintf("%03d", serialCounter*2+1)
+	clientSerial := fmt.Sprintf("%03d", serialCounter*2+2)
+
+	serialCounter = serialCounter + 1
+
+	serverName := fmt.Sprintf("server-%s", serverSerial)
+	serverCACommonName := fmt.Sprintf("Server %s CA", serverSerial)
+	serverCertName := fmt.Sprintf("server-instance-%s", serverSerial)
+	serverCertCommonName := fmt.Sprintf("server%s.example.com", serverSerial)
+
+	clientName := fmt.Sprintf("clients-%s", serverSerial)
+	clientCACommonName := fmt.Sprintf("Clients %s CA", serverSerial)
+	clientCertName := fmt.Sprintf("client-instance-%s", serverSerial)
+	clientCertCommonName := fmt.Sprintf("Client Instance %s", serverSerial)
+
+	CreateSignedCert(root, CA, serverSerial, serverName, serverCACommonName)
+	CreateSignedCert(root, serverName, serverSerial, serverCertName, serverCertCommonName)
+
+	CreateSignedCert(root, CA, clientSerial, clientName, clientCACommonName)
+	CreateSignedCert(root, clientName, serverSerial, clientCertName, clientCertCommonName)
+
+	return clientServerKeyPairs{
+		serverCert: path.Join(root, fmt.Sprintf("%s-cert.pem", serverCertName)),
+		serverKey:  path.Join(root, fmt.Sprintf("%s-key.pem", serverCertName)),
+		serverCA:   path.Join(root, fmt.Sprintf("%s-cert.pem", serverName)),
+		clientCert: path.Join(root, fmt.Sprintf("%s-cert.pem", clientCertName)),
+		clientKey:  path.Join(root, fmt.Sprintf("%s-key.pem", clientCertName)),
+		clientCA:   path.Join(root, fmt.Sprintf("%s-cert.pem", clientName)),
+		serverName: serverCertCommonName,
+	}
+
+}
+
+func getServerConfig(keypairs clientServerKeyPairs) (*tls.Config, error) {
 	return vttls.ServerConfig(
-		keypairs.ClientCert,
-		keypairs.ClientKey,
-		keypairs.ServerCA)
+		keypairs.clientCert,
+		keypairs.clientKey,
+		keypairs.serverCA)
 }
 
-func getClientConfig(keypairs ClientServerKeyPairs) (*tls.Config, error) {
+func getClientConfig(keypairs clientServerKeyPairs) (*tls.Config, error) {
 	return vttls.ClientConfig(
-		keypairs.ClientCert,
-		keypairs.ClientKey,
-		keypairs.ServerCA,
-		keypairs.ServerName)
+		keypairs.clientCert,
+		keypairs.clientKey,
+		keypairs.serverCA,
+		keypairs.serverName)
 }
 
 func TestServerTLSConfigCaching(t *testing.T) {
@@ -191,7 +242,7 @@ func TestClientTLSConfigCaching(t *testing.T) {
 	})
 }
 
-func testConfigGeneration(t *testing.T, rootPrefix string, generateConfig func(ClientServerKeyPairs) (*tls.Config, error), getCertPool func(tlsConfig *tls.Config) *x509.CertPool) {
+func testConfigGeneration(t *testing.T, rootPrefix string, generateConfig func(clientServerKeyPairs) (*tls.Config, error), getCertPool func(tlsConfig *tls.Config) *x509.CertPool) {
 	// Our test root.
 	root, err := ioutil.TempDir("", rootPrefix)
 	if err != nil {
@@ -201,8 +252,8 @@ func testConfigGeneration(t *testing.T, rootPrefix string, generateConfig func(C
 
 	const configsToGenerate = 1
 
-	firstClientServerKeyPairs := CreateClientServerCertPairs(root)
-	secondClientServerKeyPairs := CreateClientServerCertPairs(root)
+	firstClientServerKeyPairs := createClientServerCertPairs(root)
+	secondClientServerKeyPairs := createClientServerCertPairs(root)
 
 	firstExpectedConfig, _ := generateConfig(firstClientServerKeyPairs)
 	secondExpectedConfig, _ := generateConfig(secondClientServerKeyPairs)
diff --git a/go/vt/topo/etcd2topo/server_test.go b/go/vt/topo/etcd2topo/server_test.go
index 7e2a0784b9..c782e9638a 100644
--- a/go/vt/topo/etcd2topo/server_test.go
+++ b/go/vt/topo/etcd2topo/server_test.go
@@ -25,10 +25,6 @@ import (
 	"testing"
 	"time"
 
-	"vitess.io/vitess/go/vt/tlstest"
-
-	"github.com/coreos/etcd/pkg/transport"
-
 	"golang.org/x/net/context"
 
 	"github.com/coreos/etcd/clientv3"
@@ -93,130 +89,6 @@ func startEtcd(t *testing.T) (*exec.Cmd, string, string) {
 	return cmd, dataDir, clientAddr
 }
 
-// startEtcdWithTLS starts an etcd subprocess with TLS setup, and waits for it to be ready.
-func startEtcdWithTLS(t *testing.T) (string, *tlstest.ClientServerKeyPairs, func()) {
-	// Create a temporary directory.
-	dataDir, err := ioutil.TempDir("", "etcd")
-	if err != nil {
-		t.Fatalf("cannot create tempdir: %v", err)
-	}
-
-	// Get our two ports to listen to.
-	port := testfiles.GoVtTopoEtcd2topoPort
-	name := "vitess_unit_test"
-	clientAddr := fmt.Sprintf("https://localhost:%v", port+2)
-	peerAddr := fmt.Sprintf("https://localhost:%v", port+3)
-	initialCluster := fmt.Sprintf("%v=%v", name, peerAddr)
-
-	certs := tlstest.CreateClientServerCertPairs(dataDir)
-
-	cmd := exec.Command("etcd",
-		"-name", name,
-		"-advertise-client-urls", clientAddr,
-		"-initial-advertise-peer-urls", peerAddr,
-		"-listen-client-urls", clientAddr,
-		"-listen-peer-urls", peerAddr,
-		"-initial-cluster", initialCluster,
-		"-cert-file", certs.ServerCert,
-		"-key-file", certs.ServerKey,
-		"-trusted-ca-file", certs.ClientCA,
-		"-peer-trusted-ca-file", certs.ClientCA,
-		"-peer-cert-file", certs.ServerCert,
-		"-peer-key-file", certs.ServerKey,
-		"-client-cert-auth",
-		"-data-dir", dataDir)
-
-	cmd.Stderr = os.Stderr
-	cmd.Stdout = os.Stdout
-	err = cmd.Start()
-	if err != nil {
-		t.Fatalf("failed to start etcd: %v", err)
-	}
-
-	// Safe now to build up TLS info.
-	tlsInfo := transport.TLSInfo{
-		CertFile:      certs.ClientCert,
-		KeyFile:       certs.ClientKey,
-		TrustedCAFile: certs.ServerCA,
-	}
-
-	tlsConfig, err := tlsInfo.ClientConfig()
-
-	var cli *clientv3.Client
-	// Create client
-	start := time.Now()
-	for {
-		// Create a client to connect to the created etcd.
-		cli, err = clientv3.New(clientv3.Config{
-			Endpoints:   []string{clientAddr},
-			TLS:         tlsConfig,
-			DialTimeout: 5 * time.Second,
-		})
-		if err == nil {
-			break
-		}
-		t.Logf("error establishing client for etcd tls test: %v", err)
-		if time.Since(start) > 60*time.Second {
-			t.Fatalf("failed to start client for etcd tls test in time")
-		}
-		time.Sleep(100 * time.Millisecond)
-	}
-	defer cli.Close()
-
-	// Wait until we can list "/", or timeout.
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-	start = time.Now()
-	for {
-		if _, err := cli.Get(ctx, "/"); err == nil {
-			break
-		}
-		if time.Since(start) > 60*time.Second {
-			t.Fatalf("failed to start etcd daemon in time")
-		}
-		time.Sleep(100 * time.Millisecond)
-	}
-
-	stopEtcd := func() {
-		cmd.Process.Kill()
-		cmd.Wait()
-		os.RemoveAll(dataDir)
-	}
-
-	return clientAddr, &certs, stopEtcd
-}
-
-func TestEtcd2TLS(t *testing.T) {
-	// Start a single etcd in the background.
-	clientAddr, certs, stopEtcd := startEtcdWithTLS(t)
-	defer stopEtcd()
-
-	testIndex := 0
-	testRoot := fmt.Sprintf("/test-%v", testIndex)
-
-	// Create the server on the new root.
-	server, err := NewServerWithOpts(clientAddr, testRoot, certs.ClientCert, certs.ClientKey, certs.ServerCA)
-	if err != nil {
-		t.Fatalf("NewServerWithOpts failed: %v", err)
-	}
-	defer server.Close()
-
-	testCtx := context.Background()
-	testKey := "testkey"
-	testVal := "testval"
-	_, err = server.Create(testCtx, testKey, []byte(testVal))
-	if err != nil {
-		t.Fatalf("Failed to set key value pair: %v", err)
-	}
-	val, _, err := server.Get(testCtx, testKey)
-	if err != nil {
-		t.Fatalf("Failed to retrieve value at key we just set: %v", err)
-	}
-	if string(val) != testVal {
-		t.Fatalf("Value returned doesn't match %s, err: %v", testVal, err)
-	}
-}
-
 func TestEtcd2Topo(t *testing.T) {
 	// Start a single etcd in the background.
 	cmd, dataDir, clientAddr := startEtcd(t)
diff --git a/go/vt/vtgate/executor_test.go b/go/vt/vtgate/executor_test.go
index 8a42e884bc..8767cd0928 100644
--- a/go/vt/vtgate/executor_test.go
+++ b/go/vt/vtgate/executor_test.go
@@ -890,13 +890,6 @@ func TestExecutorShow(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-
-	// Test SHOW FULL COLUMNS FROM where query has a qualifier
-	_, err = executor.Execute(context.Background(), "TestExecute", session, fmt.Sprintf("show full columns from %v.table1", KsTestUnsharded), nil)
-	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
-	}
-
 	// Just test for first & last.
 	qr.Rows = [][]sqltypes.Value{qr.Rows[0], qr.Rows[len(qr.Rows)-1]}
 	wantqr = &sqltypes.Result{
diff --git a/go/vt/vtgate/planbuilder/testdata/filter_cases.txt b/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
index 53fae69ba0..0130fe08b7 100644
--- a/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
@@ -233,17 +233,13 @@
 {
   "Original": "select user_extra.id from user left join user_extra on user.id = user_extra.user_id where user_extra.user_id = 5",
   "Instructions": {
-    "Opcode": "SelectEqualUnique",
+    "Opcode": "SelectScatter",
     "Keyspace": {
       "Name": "user",
       "Sharded": true
     },
     "Query": "select user_extra.id from user left join user_extra on user.id = user_extra.user_id where user_extra.user_id = 5",
     "FieldQuery": "select user_extra.id from user left join user_extra on user.id = user_extra.user_id where 1 != 1",
-    "Vindex": "user_index",
-    "Values": [
-      5
-    ],
     "Table": "user"
   }
 }
diff --git a/go/vt/vtgate/planbuilder/testdata/from_cases.txt b/go/vt/vtgate/planbuilder/testdata/from_cases.txt
index 67943ac9df..6428be6a65 100644
--- a/go/vt/vtgate/planbuilder/testdata/from_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/from_cases.txt
@@ -927,19 +927,35 @@
   }
 }
 
-# reference table can merge with other opcodes left to right.
+# reference table doesn't merge with other opcodes yet.
 "select ref.col from ref join user"
 {
   "Original": "select ref.col from ref join user",
   "Instructions": {
-    "Opcode": "SelectScatter",
-    "Keyspace": {
-      "Name": "user",
-      "Sharded": true
+    "Opcode": "Join",
+    "Left": {
+      "Opcode": "SelectReference",
+      "Keyspace": {
+        "Name": "user",
+        "Sharded": true
+      },
+      "Query": "select ref.col from ref",
+      "FieldQuery": "select ref.col from ref where 1 != 1",
+      "Table": "ref"
     },
-    "Query": "select ref.col from ref join user",
-    "FieldQuery": "select ref.col from ref join user where 1 != 1",
-    "Table": "user"
+    "Right": {
+      "Opcode": "SelectScatter",
+      "Keyspace": {
+        "Name": "user",
+        "Sharded": true
+      },
+      "Query": "select 1 from user",
+      "FieldQuery": "select 1 from user where 1 != 1",
+      "Table": "user"
+    },
+    "Cols": [
+      -1
+    ]
   }
 }
 
diff --git a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
index 67a05f8aee..77cee7da74 100644
--- a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
@@ -432,3 +432,8 @@
 
 "select func(keyspace_id) from user_index where id = :id"
 "unsupported: expression on results of a vindex function"
+
+# Multi-table unique vindex constraint on left table of left join, two levels of join, simple aggregation
+# This should work, but doesn't. See https://github.com/vitessio/vitess/issues/4772
+"select user.id, count(*) from user left join user_extra ue1 on user.id = ue1.user_id left join user_extra ue2 on ue1.user_id = ue2.user_id group by user.id"
+"unsupported: cross-shard query with aggregates"
diff --git a/go/vt/vttablet/tabletserver/vstreamer/engine_test.go b/go/vt/vttablet/tabletserver/vstreamer/engine_test.go
index 28c5da5643..b45e240265 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/engine_test.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/engine_test.go
@@ -100,36 +100,46 @@ func TestUpdateVSchema(t *testing.T) {
 	expectUpdateCount(t, startCount+1)
 
 	want := `{
-  "sharded": true,
-  "tables": {
-    "t1": {
-      "name": "t1",
-      "column_vindexes": [
-        {
-          "columns": [
-            "id1"
-          ],
-          "type": "hash",
-          "name": "hash",
-          "vindex": {}
-        }
-      ],
-      "ordered": [
-        {
-          "columns": [
-            "id1"
+  "routing_rules": {},
+  "keyspaces": {
+    "vttest": {
+      "sharded": true,
+      "tables": {
+        "dual": {
+          "type": "reference",
+          "name": "dual"
+        },
+        "t1": {
+          "name": "t1",
+          "column_vindexes": [
+            {
+              "columns": [
+                "id1"
+              ],
+              "type": "hash",
+              "name": "hash",
+              "vindex": {}
+            }
           ],
-          "type": "hash",
-          "name": "hash",
-          "vindex": {}
+          "ordered": [
+            {
+              "columns": [
+                "id1"
+              ],
+              "type": "hash",
+              "name": "hash",
+              "vindex": {}
+            }
+          ]
         }
-      ]
+      },
+      "vindexes": {
+        "hash": {}
+      }
     }
-  },
-  "vindexes": {
-    "hash": {}
   }
 }`
+
 	b, err := json.MarshalIndent(engine.vschema(), "", "  ")
 	if err != nil {
 		t.Fatal(err)
diff --git a/go/vt/vttablet/tabletserver/vstreamer/local_vschema_test.go b/go/vt/vttablet/tabletserver/vstreamer/local_vschema_test.go
new file mode 100644
index 0000000000..0ac313eb77
--- /dev/null
+++ b/go/vt/vttablet/tabletserver/vstreamer/local_vschema_test.go
@@ -0,0 +1,142 @@
+/*
+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 vstreamer
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	vschemapb "vitess.io/vitess/go/vt/proto/vschema"
+	"vitess.io/vitess/go/vt/vtgate/vindexes"
+)
+
+var testSrvVSchema = &vschemapb.SrvVSchema{
+	Keyspaces: map[string]*vschemapb.Keyspace{
+		"ks1": {
+			Sharded: true,
+			Vindexes: map[string]*vschemapb.Vindex{
+				"duphash": {
+					Type: "hash",
+				},
+			},
+			Tables: map[string]*vschemapb.Table{
+				"t1": {
+					ColumnVindexes: []*vschemapb.ColumnVindex{{
+						Name:    "duphash",
+						Columns: []string{"id"},
+					}},
+				},
+			},
+		},
+		"ks2": {
+			Sharded: true,
+			Vindexes: map[string]*vschemapb.Vindex{
+				"duphash": {
+					Type: "hash",
+				},
+				"otherhash": {
+					Type: "hash",
+				},
+			},
+		},
+	},
+}
+
+func TestFindTable(t *testing.T) {
+	vschema, err := vindexes.BuildVSchema(testSrvVSchema)
+	require.NoError(t, err)
+
+	testcases := []struct {
+		keyspace  string
+		tablename string
+		err       string
+	}{{
+		keyspace:  "ks1",
+		tablename: "t1",
+		err:       "",
+	}, {
+		keyspace:  "ks1",
+		tablename: "t2",
+		err:       "table t2 not found",
+	}, {
+		keyspace:  "noks",
+		tablename: "t2",
+		err:       "keyspace noks not found in vschema",
+	}}
+	for _, tcase := range testcases {
+		lvs := &localVSchema{
+			keyspace: tcase.keyspace,
+			vschema:  vschema,
+		}
+		table, err := lvs.FindTable(tcase.tablename)
+		if err != nil {
+			assert.EqualError(t, err, tcase.err, tcase.keyspace, tcase.tablename)
+			continue
+		}
+		assert.NoError(t, err, tcase.keyspace, tcase.tablename)
+		assert.Equal(t, table.Name.String(), tcase.tablename, tcase.keyspace, tcase.tablename)
+	}
+}
+
+func TestFindOrCreateVindex(t *testing.T) {
+	vschema, err := vindexes.BuildVSchema(testSrvVSchema)
+	require.NoError(t, err)
+
+	testcases := []struct {
+		name string
+		err  string
+	}{{
+		name: "otherhash",
+		err:  "",
+	}, {
+		name: "ks1.duphash",
+		err:  "",
+	}, {
+		name: "hash",
+		err:  "",
+	}, {
+		name: "a.b.c",
+		err:  "invalid vindex name: a.b.c",
+	}, {
+		name: "duphash",
+		err:  "ambiguous vindex reference: duphash",
+	}, {
+		name: "ks1.hash",
+		err:  "vindex ks1.hash not found",
+	}, {
+		name: "none",
+		err:  `vindexType "none" not found`,
+	}}
+	for _, tcase := range testcases {
+		lvs := &localVSchema{
+			keyspace: "ks1",
+			vschema:  vschema,
+		}
+		vindex, err := lvs.FindOrCreateVindex(tcase.name)
+		if err != nil {
+			assert.EqualError(t, err, tcase.err, tcase.name)
+			continue
+		}
+		assert.NoError(t, err, tcase.name)
+		splits := strings.Split(tcase.name, ".")
+		want := splits[len(splits)-1]
+		assert.Equal(t, vindex.String(), want, tcase.name)
+	}
+}
diff --git a/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go b/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go
index e6d38059e9..0ff9bfea5f 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go
@@ -32,7 +32,7 @@ import (
 	vschemapb "vitess.io/vitess/go/vt/proto/vschema"
 )
 
-var testKSChema *vindexes.KeyspaceSchema
+var testLocalVSchema *localVSchema
 
 func init() {
 	input := `{
@@ -74,11 +74,19 @@ func init() {
 	if err := json2.Unmarshal([]byte(input), &kspb); err != nil {
 		panic(fmt.Errorf("Unmarshal failed: %v", err))
 	}
-	kschema, err := vindexes.BuildKeyspaceSchema(&kspb, "ks")
+	srvVSchema := &vschemapb.SrvVSchema{
+		Keyspaces: map[string]*vschemapb.Keyspace{
+			"ks": &kspb,
+		},
+	}
+	vschema, err := vindexes.BuildVSchema(srvVSchema)
 	if err != nil {
 		panic(err)
 	}
-	testKSChema = kschema
+	testLocalVSchema = &localVSchema{
+		keyspace: "ks",
+		vschema:  vschema,
+	}
 }
 
 func TestMustSendDDL(t *testing.T) {
@@ -339,7 +347,7 @@ func TestPlanbuilder(t *testing.T) {
 				Type:   sqltypes.Int64,
 			}, {
 				Alias:         sqlparser.NewColIdent("keyspace_id"),
-				Vindex:        testKSChema.Vindexes["region_vdx"],
+				Vindex:        testLocalVSchema.vschema.Keyspaces["ks"].Vindexes["region_vdx"],
 				VindexColumns: []int{0, 1},
 				Type:          sqltypes.VarBinary,
 			}},
@@ -351,7 +359,7 @@ func TestPlanbuilder(t *testing.T) {
 	}, {
 		inTable: t2,
 		inRule:  &binlogdatapb.Rule{Match: "/.*/", Filter: "-80"},
-		outErr:  `no vschema definition for table t2`,
+		outErr:  `table t2 not found`,
 	}, {
 		inTable: t1alt,
 		inRule:  &binlogdatapb.Rule{Match: "/.*/", Filter: "-80"},
@@ -457,7 +465,7 @@ func TestPlanbuilder(t *testing.T) {
 	}}
 
 	for _, tcase := range testcases {
-		plan, err := buildPlan(tcase.inTable, testKSChema, &binlogdatapb.Filter{
+		plan, err := buildPlan(tcase.inTable, testLocalVSchema, &binlogdatapb.Filter{
 			Rules: []*binlogdatapb.Rule{tcase.inRule},
 		})
 		if plan != nil {
diff --git a/tools/unit_test_runner.sh b/tools/unit_test_runner.sh
index 879626d7bf..4cffb3eb91 100755
--- a/tools/unit_test_runner.sh
+++ b/tools/unit_test_runner.sh
@@ -55,8 +55,6 @@ if [ $? -ne 0 ]; then
   exit 1
 fi
 
-echo '# Flaky tests (3 attempts permitted)'
-
 # Run flaky tests sequentially. Retry when necessary.
 for pkg in $flaky_tests; do
   max_attempts=3