Repository: vitessio/vitess
Pull request #8503 (2021-07-20 14:27:19)

Refactor authentication server plugin mechanism
This change implements a large refactor of the plugin authentication mechanism. This refactor needed for a number of reasons:

- The current AuthServer interface is not well designed. Many of the AuthServer implementations implement at least one if not multiple methods as a panic() as they are unused or they implement something never really called (like the salt generation). This indicates that the current interface doesn't abstract things well.
- Adding `caching_sha2_password` as a server side mechanism would make this interface problem worse. It would add even more methods that would have emply implementations / panic().
- It is impossible right now for a plugin to support multiple authentication mechanisms. The caching_sha2_password plugin allows for caching of passwords and could be used to improve for example the LDAP  plugin. But we can't then still keep the `mysql_clear_password` as a fallback for such an improvement in the LDAP plugin.

Based on these issues, I'm proposing this refactor. It allows for a number of things that also make auth server implementations easier.

- An auth server can implement multiple auth methods. This allows for the above mentioned optimization for the LDAP plugin and many other improvements. Note that these improvements are not implemented here, but it opens up the opportunity.
- There are helpers for all the already supported authentication mechanisms plus basic support for caching_sha2_password. The last only supports TLS connections or Unix sockets right now, since the full authentication protocol is much complicated. The value of the latter is also debatable in the context of Vitess. It depends on a TLS certificate and private key also for the plain text full auth cycle and in that case, TLS would already be available as well anyway. For clients that want caching_sha2_password, switching those to enable TLS seems like a much simpler solution.
- With the helpers, specific interfaces are provided that plugins can implement. Those interfaces provide a simpler way than the complex Negotiate from before which might need to read packets etc. For these helpers, all reading / writing of the protocol is already dealt with and not something auth server plugin writers need to be concerned with.

Given the above advantages, I think the current setup is a better design and better fits what we'd want to use for authentication plugins.

Lastly, this does break any existing plugin implementations that are written in Go and that people might have done themselves in their own Vitess forks etc. I don't know if that's an expected interface / something that needs a larger announcement.

The functionality of all the built in plugins is exactly the same. They have been switched to the new internal APIs, but the external functionality, config files, configuration flags etc. have all stayed exactly the same. 

## Related Issue(s)

https://github.com/vitessio/vitess/issues/5399 is related, but that was closed after only client side support for `caching_sha2_password` was implemented.

With the refactors here, it also becomes possible to implement `caching_sha2_password` for the server side component in Vitess.

## Checklist
- [x] Tests were added or are not required
- [x] Documentation was added or is not required

Not really sure about the documentation. I don't think we document the internal Go interfaces, only things like configuration flags etc. for plugins that all stay the same? 


Patch:
diff --git a/go/mysql/auth_server.go b/go/mysql/auth_server.go
index b568537e16..e86f08a66c 100644
--- a/go/mysql/auth_server.go
+++ b/go/mysql/auth_server.go
@@ -17,14 +17,15 @@ limitations under the License.
 package mysql
 
 import (
-	"bytes"
 	"crypto/rand"
 	"crypto/rsa"
 	"crypto/sha1"
 	"crypto/sha256"
+	"crypto/subtle"
+	"crypto/x509"
 	"encoding/hex"
 	"net"
-	"strings"
+	"sync"
 
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/proto/vtrpc"
@@ -32,94 +33,217 @@ import (
 )
 
 // AuthServer is the interface that servers must implement to validate
-// users and passwords. It has two modes:
+// users and passwords. It needs to be able to return a list of AuthMethod
+// interfaces which implement the supported auth methods for the server.
+type AuthServer interface {
+	// AuthMethods returns a list of auth methods that are part of this
+	// interface. Building an authentication server usually means
+	// creating AuthMethod instances with the known helpers for the
+	// currently supported AuthMethod implementations.
+	//
+	// When a client connects, the server checks the list of auth methods
+	// available. If an auth method for the requested auth mechanism by the
+	// client is available, it will be used immediately.
+	//
+	// If there is no overlap between the provided auth methods and
+	// the one the client requests, the server will send back an
+	// auth switch request using the first provided AuthMethod in this list.
+	AuthMethods() []AuthMethod
+
+	// DefaultAuthMethodDescription returns the auth method that the auth server
+	// sends during the initial server handshake. This needs to be either
+	// `mysql_native_password` or `caching_sha2_password` as those are the only
+	// supported auth methods during the initial handshake.
+	//
+	// It's not needed to also support those methods in the AuthMethods(),
+	// in fact, if you want to only support for example clear text passwords,
+	// you must still return `mysql_native_password` or `caching_sha2_password`
+	// here and the auth switch protocol will be used to switch to clear text.
+	DefaultAuthMethodDescription() AuthMethodDescription
+}
+
+// AuthMethod interface for concrete auth method implementations.
+// When building an auth server, you usually don't implement these yourself
+// but the helper methods to build AuthMethod instances should be used.
+type AuthMethod interface {
+	// Name returns the auth method description for this implementation.
+	// This is the name that is sent as the auth plugin name during the
+	// Mysql authentication protocol handshake.
+	Name() AuthMethodDescription
+
+	// HandleUser verifies if the current auth method can authenticate
+	// the given user with the current auth method. This can be useful
+	// for example if you only have a plain text of hashed password
+	// for specific users and not all users and auth method support
+	// depends on what you have.
+	HandleUser(conn *Conn, user string) bool
+
+	// AuthPluginData generates the information for the auth plugin.
+	// This is included in for example the auth switch request. This
+	// is auth plugin specific and opaque to the Mysql handshake
+	// protocol.
+	AuthPluginData() ([]byte, error)
+
+	// HandleAuthPluginData handles the returned auth plugin data from
+	// the client. The original data the server sent is also included
+	// which can include things like the salt for `mysql_native_password`.
+	//
+	// The remote address is provided for plugins that also want to
+	// do additional checks like IP based restrictions.
+	HandleAuthPluginData(conn *Conn, user string, serverAuthPluginData []byte, clientAuthPluginData []byte, remoteAddr net.Addr) (Getter, error)
+}
+
+// UserValidator is an interface that allows checking if a specific
+// user will work for an auth method. This interface is called by
+// all the default helpers that create AuthMethod instances for
+// the various supported Mysql authentication methods.
+type UserValidator interface {
+	HandleUser(user string) bool
+}
+
+// CacheState is a state that is returned by the UserEntryWithCacheHash
+// method from the CachingStorage interface. This state is needed to indicate
+// whether the authentication is accepted, rejected by the cache itself
+// or if the cache can't fullfill the request. In that case it indicates
+// that with AuthNeedMoreData.
+type CacheState int
+
+const (
+	// AuthRejected is used when the cache knows the request can be rejected.
+	AuthRejected CacheState = iota
+	// AuthAccepted is used when the cache knows the request can be accepted.
+	AuthAccepted
+	// AuthNeedMoreData is used when the cache doesn't know the answer and more data is needed.
+	AuthNeedMoreData
+)
+
+// HashStorage describes an object that is suitable to retrieve user information
+// based on the hashed authentication response for mysql_native_password.
 //
-// 1. using salt the way MySQL native auth does it. In that case, the
-// password is not sent in the clear, but the salt is used to hash the
-// password both on the client and server side, and the result is sent
-// and compared.
+// In general, an implementation of this would use an internally stored password
+// that is hashed twice with SHA1.
 //
-// 2. sending the user / password in the clear (using MySQL Cleartext
-// method). The server then gets access to both user and password, and
-// can authenticate using any method. If SSL is not used, it means the
-// password is sent in the clear. That may not be suitable for some
-// use cases.
-type AuthServer interface {
-	// AuthMethod returns the authentication method to use for the
-	// given user. If this returns MysqlNativePassword
-	// (mysql_native_password), then ValidateHash() will be
-	// called, and no further roundtrip with the client is
-	// expected. If anything else is returned, Negotiate()
-	// will be called on the connection, and the AuthServer
-	// needs to handle the packets.
-	AuthMethod(user string) (string, error)
-
-	// Salt returns the salt to use for a connection.
-	// It should be 20 bytes of data.
-	// Most implementations should just use mysql.NewSalt().
-	// (this is meant to support a plugin that would use an
-	// existing MySQL server as the source of auth, and just forward
-	// the salt generated by that server).
-	// Do not return zero bytes, as a known salt can be the source
-	// of a crypto attack.
-	Salt() ([]byte, error)
-
-	// ValidateHash validates the data sent by the client matches
-	// what the server computes.  It also returns the user data.
-	ValidateHash(salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, error)
-
-	// Negotiate is called if AuthMethod returns anything else
-	// than MysqlNativePassword. It is handed the connection after the
-	// AuthSwitchRequest packet is sent.
-	// - If the negotiation fails, it should just return an error
-	// (should be a SQLError if possible).
-	// The framework is responsible for writing the Error packet
-	// and closing the connection in that case.
-	// - If the negotiation works, it should return the Getter,
-	// and no error. The framework is responsible for writing the
-	// OK packet.
-	Negotiate(c *Conn, user string, remoteAddr net.Addr) (Getter, error)
+// The VerifyHashedMysqlNativePassword helper method can be used to verify
+// such a hash based on the salt and auth response provided here after retrieving
+// the hashed password from the storage.
+type HashStorage interface {
+	UserEntryWithHash(userCerts []*x509.Certificate, salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, error)
 }
 
-// authServers is a registry of AuthServer implementations.
-var authServers = make(map[string]AuthServer)
+// PlainTextStorage describes an object that is suitable to retrieve user information
+// based on the plain text password of a user. This can be obtained through various
+// Mysql authentication methods, such as `mysql_clear_passwrd`, `dialog` or
+// `caching_sha2_password` in the full authentication handshake case of the latter.
+//
+// This mechanism also would allow for picking your own password storage in the backend,
+// such as BCrypt, SCrypt, PBKDF2 or Argon2 once the plain text is obtained.
+//
+// When comparing plain text passwords directly, please ensure to use `subtle.ConstantTimeCompare`
+// to prevent timing based attacks on the password.
+type PlainTextStorage interface {
+	UserEntryWithPassword(userCerts []*x509.Certificate, user string, password string, remoteAddr net.Addr) (Getter, error)
+}
 
-// RegisterAuthServerImpl registers an implementations of AuthServer.
-func RegisterAuthServerImpl(name string, authServer AuthServer) {
-	if _, ok := authServers[name]; ok {
-		log.Fatalf("AuthServer named %v already exists", name)
+// CachingStorage describes an object that is suitable to retrieve user information
+// based on a hashed value of the password. This applies to the `caching_sha2_password`
+// authentication method.
+//
+// The cache would hash the password internally as `SHA256(SHA256(password))`.
+//
+// The VerifyHashedCachingSha2Password helper method can be used to verify
+// such a hash based on the salt and auth response provided here after retrieving
+// the hashed password from the cache.
+type CachingStorage interface {
+	UserEntryWithCacheHash(userCerts []*x509.Certificate, salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, CacheState, error)
+}
+
+// NewMysqlNativeAuthMethod will create a new AuthMethod that implements the
+// `mysql_native_password` handshake. The caller will need to provide a storage
+// object and validator that will be called during the handshake phase.
+func NewMysqlNativeAuthMethod(layer HashStorage, validator UserValidator) AuthMethod {
+	authMethod := mysqlNativePasswordAuthMethod{
+		storage:   layer,
+		validator: validator,
 	}
-	authServers[name] = authServer
+	return &authMethod
 }
 
-// GetAuthServer returns an AuthServer by name, or log.Exitf.
-func GetAuthServer(name string) AuthServer {
-	authServer, ok := authServers[name]
-	if !ok {
-		log.Exitf("no AuthServer name %v registered", name)
+// NewMysqlClearAuthMethod will create a new AuthMethod that implements the
+// `mysql_clear_password` handshake. The caller will need to provide a storage
+// object for plain text passwords and validator that will be called during the
+// handshake phase.
+func NewMysqlClearAuthMethod(layer PlainTextStorage, validator UserValidator) AuthMethod {
+	authMethod := mysqlClearAuthMethod{
+		storage:   layer,
+		validator: validator,
 	}
-	return authServer
+	return &authMethod
 }
 
-// NewSalt returns a 20 character salt.
-func NewSalt() ([]byte, error) {
-	salt := make([]byte, 20)
-	if _, err := rand.Read(salt); err != nil {
-		return nil, err
+// Constants for the dialog plugin.
+const (
+	// Default message if no custom message
+	// is configured. This is used when the message
+	// is the empty string.
+	mysqlDialogDefaultMessage = "Enter password: "
+
+	// Dialog plugin is similar to clear text, but can respond to multiple
+	// prompts in a row. This is not yet implemented.
+	// Follow questions should be prepended with a `cmd` byte:
+	// 0x02 - ordinary question
+	// 0x03 - last question
+	// 0x04 - password question
+	// 0x05 - last password
+	mysqlDialogAskPassword = 0x04
+)
+
+// NewMysqlDialogAuthMethod will create a new AuthMethod that implements the
+// `dialog` handshake. The caller will need to provide a storage object for plain
+// text passwords and validator that will be called during the handshake phase.
+// The message given will be sent as part of the dialog. If the empty string is
+// provided, the default message of "Enter password: " will be used.
+func NewMysqlDialogAuthMethod(layer PlainTextStorage, validator UserValidator, msg string) AuthMethod {
+	if msg == "" {
+		msg = mysqlDialogDefaultMessage
 	}
 
-	// Salt must be a legal UTF8 string.
-	for i := 0; i < len(salt); i++ {
-		salt[i] &= 0x7f
-		if salt[i] == '\x00' || salt[i] == '$' {
-			salt[i]++
-		}
+	authMethod := mysqlDialogAuthMethod{
+		storage:   layer,
+		validator: validator,
+		msg:       msg,
 	}
+	return &authMethod
+}
 
-	return salt, nil
+// NewSha2CachingAuthMethod will create a new AuthMethod that implements the
+// `caching_sha2_password` handshake. The caller will need to provide a cache
+// object for the fast auth path and a plain text storage object that will
+// be called if the return of the first layer indicates the full auth dance is
+// needed.
+//
+// Right now we only support caching_sha2_password over TLS or a Unix socket.
+//
+// If TLS is not enabled, the client needs to encrypt it with the public
+// key of the server. In that case, Vitess is already configured with
+// a certificate anyway, so we recommend to use TLS if you want to use
+// caching_sha2_password in that case instead of allowing the plain
+// text fallback path here.
+//
+// This might change in the future if there's a good argument and implementation
+// for allowing the plain text path here as well.
+func NewSha2CachingAuthMethod(layer1 CachingStorage, layer2 PlainTextStorage, validator UserValidator) AuthMethod {
+	authMethod := mysqlCachingSha2AuthMethod{
+		cache:     layer1,
+		storage:   layer2,
+		validator: validator,
+	}
+	return &authMethod
 }
 
 // ScrambleMysqlNativePassword computes the hash of the password using 4.1+ method.
+//
+// This can be used for example inside a `mysql_native_password` plugin implementation
+// if the backend storage implements storage of plain text passwords.
 func ScrambleMysqlNativePassword(salt, password []byte) []byte {
 	if len(password) == 0 {
 		return nil
@@ -148,34 +272,63 @@ func ScrambleMysqlNativePassword(salt, password []byte) []byte {
 	return scramble
 }
 
-func isPassScrambleMysqlNativePassword(reply, salt []byte, mysqlNativePassword string) bool {
-	/*
-		SERVER:  recv(reply)
-				 hash_stage1=xor(reply, sha1(salt,hash))
-				 candidate_hash2=sha1(hash_stage1)
-				 check(candidate_hash2==hash)
-	*/
-	if len(reply) == 0 {
-		return false
+// DecodeMysqlNativePasswordHex decodes the standard format used by MySQL
+// for 4.1 style password hashes. It drops the optionally leading * before
+// decoding the rest as a hex encoded string.
+func DecodeMysqlNativePasswordHex(hexEncodedPassword string) ([]byte, error) {
+	if hexEncodedPassword[0] == '*' {
+		hexEncodedPassword = hexEncodedPassword[1:]
 	}
+	return hex.DecodeString(hexEncodedPassword)
+}
 
-	if mysqlNativePassword == "" {
+// VerifyHashedMysqlNativePassword verifies a client reply against a stored hash.
+//
+// This can be used for example inside a `mysql_native_password` plugin implementation
+// if the backend storage where the stored password is a SHA1(SHA1(password)).
+//
+// All values here are non encoded byte slices, so if you store for example the double
+// SHA1 of the password as hex encoded characters, you need to decode that first.
+// See DecodeMysqlNativePasswordHex for a decoding helper for the standard encoding
+// format of this hash used by MySQL.
+func VerifyHashedMysqlNativePassword(reply, salt, hashedNativePassword []byte) bool {
+	if len(reply) == 0 || len(hashedNativePassword) == 0 {
 		return false
 	}
 
-	if strings.Contains(mysqlNativePassword, "*") {
-		mysqlNativePassword = mysqlNativePassword[1:]
+	// scramble = SHA1(salt+hash)
+	crypt := sha1.New()
+	crypt.Write(salt)
+	crypt.Write(hashedNativePassword)
+	scramble := crypt.Sum(nil)
+
+	for i := range scramble {
+		scramble[i] ^= reply[i]
 	}
+	hashStage1 := scramble
 
-	hash, err := hex.DecodeString(mysqlNativePassword)
-	if err != nil {
+	crypt.Reset()
+	crypt.Write(hashStage1)
+	candidateHash2 := crypt.Sum(nil)
+
+	return subtle.ConstantTimeCompare(candidateHash2, hashedNativePassword) == 1
+}
+
+// VerifyHashedCachingSha2Password verifies a client reply against a stored hash.
+//
+// This can be used for example inside a `caching_sha2_password` plugin implementation
+// if the cache storage uses password keys with SHA256(SHA256(password)).
+//
+// All values here are non encoded byte slices, so if you store for example the double
+// SHA256 of the password as hex encoded characters, you need to decode that first.
+func VerifyHashedCachingSha2Password(reply, salt, hashedCachingSha2Password []byte) bool {
+	if len(reply) == 0 || len(hashedCachingSha2Password) == 0 {
 		return false
 	}
 
-	// scramble = SHA1(salt+hash)
-	crypt := sha1.New()
+	crypt := sha256.New()
+	crypt.Write(hashedCachingSha2Password)
 	crypt.Write(salt)
-	crypt.Write(hash)
 	scramble := crypt.Sum(nil)
 
 	// token = scramble XOR stage1Hash
@@ -188,7 +341,7 @@ func isPassScrambleMysqlNativePassword(reply, salt []byte, mysqlNativePassword s
 	crypt.Write(hashStage1)
 	candidateHash2 := crypt.Sum(nil)
 
-	return bytes.Equal(candidateHash2, hash)
+	return subtle.ConstantTimeCompare(candidateHash2, hashedCachingSha2Password) == 1
 }
 
 // ScrambleCachingSha2Password computes the hash of the password using SHA256 as required by
@@ -243,34 +396,215 @@ func EncryptPasswordWithPublicKey(salt []byte, password []byte, pub *rsa.PublicK
 	return enc, nil
 }
 
-// Constants for the dialog plugin.
-const (
-	mysqlDialogMessage = "Enter password: "
+type mysqlNativePasswordAuthMethod struct {
+	storage   HashStorage
+	validator UserValidator
+}
 
-	// Dialog plugin is similar to clear text, but can respond to multiple
-	// prompts in a row. This is not yet implemented.
-	// Follow questions should be prepended with a `cmd` byte:
-	// 0x02 - ordinary question
-	// 0x03 - last question
-	// 0x04 - password question
-	// 0x05 - last password
-	mysqlDialogAskPassword = 0x04
-)
+func (n *mysqlNativePasswordAuthMethod) Name() AuthMethodDescription {
+	return MysqlNativePassword
+}
+
+func (n *mysqlNativePasswordAuthMethod) HandleUser(conn *Conn, user string) bool {
+	return n.validator.HandleUser(user)
+}
 
-// authServerDialogSwitchData is a helper method to return the data
-// needed in the AuthSwitchRequest packet for the dialog plugin
-// to ask for a password.
-func authServerDialogSwitchData() []byte {
-	result := make([]byte, len(mysqlDialogMessage)+2)
+func (n *mysqlNativePasswordAuthMethod) AuthPluginData() ([]byte, error) {
+	salt, err := newSalt()
+	if err != nil {
+		return nil, err
+	}
+	return append(salt, 0), nil
+}
+
+func (n *mysqlNativePasswordAuthMethod) HandleAuthPluginData(conn *Conn, user string, serverAuthPluginData []byte, clientAuthPluginData []byte, remoteAddr net.Addr) (Getter, error) {
+	if serverAuthPluginData[len(serverAuthPluginData)-1] != 0x00 {
+		return nil, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+	}
+
+	salt := serverAuthPluginData[:len(serverAuthPluginData)-1]
+	return n.storage.UserEntryWithHash(conn.GetTLSClientCerts(), salt, user, clientAuthPluginData, remoteAddr)
+}
+
+type mysqlClearAuthMethod struct {
+	storage   PlainTextStorage
+	validator UserValidator
+}
+
+func (n *mysqlClearAuthMethod) Name() AuthMethodDescription {
+	return MysqlClearPassword
+}
+
+func (n *mysqlClearAuthMethod) HandleUser(conn *Conn, user string) bool {
+	return n.validator.HandleUser(user)
+}
+
+func (n *mysqlClearAuthMethod) AuthPluginData() ([]byte, error) {
+	return nil, nil
+}
+
+func (n *mysqlClearAuthMethod) HandleAuthPluginData(conn *Conn, user string, serverAuthPluginData []byte, clientAuthPluginData []byte, remoteAddr net.Addr) (Getter, error) {
+	password := ""
+	if len(clientAuthPluginData) > 0 {
+		password = string(clientAuthPluginData[:len(clientAuthPluginData)-1])
+	}
+
+	return n.storage.UserEntryWithPassword(conn.GetTLSClientCerts(), user, password, remoteAddr)
+}
+
+type mysqlDialogAuthMethod struct {
+	storage   PlainTextStorage
+	validator UserValidator
+	msg       string
+}
+
+func (n *mysqlDialogAuthMethod) Name() AuthMethodDescription {
+	return MysqlDialog
+}
+
+func (n *mysqlDialogAuthMethod) HandleUser(conn *Conn, user string) bool {
+	return n.validator.HandleUser(user)
+}
+
+func (n *mysqlDialogAuthMethod) AuthPluginData() ([]byte, error) {
+	result := make([]byte, len(n.msg)+2)
 	result[0] = mysqlDialogAskPassword
-	writeNullString(result, 1, mysqlDialogMessage)
-	return result
+	writeNullString(result, 1, n.msg)
+	return result, nil
+}
+
+func (n *mysqlDialogAuthMethod) HandleAuthPluginData(conn *Conn, user string, serverAuthPluginData []byte, clientAuthPluginData []byte, remoteAddr net.Addr) (Getter, error) {
+	return n.storage.UserEntryWithPassword(conn.GetTLSClientCerts(), user, string(clientAuthPluginData[:len(clientAuthPluginData)-1]), remoteAddr)
 }
 
-// AuthServerReadPacketString is a helper method to read a packet
-// as a null terminated string. It is used by the mysql_clear_password
-// and dialog plugins.
-func AuthServerReadPacketString(c *Conn) (string, error) {
+type mysqlCachingSha2AuthMethod struct {
+	cache     CachingStorage
+	storage   PlainTextStorage
+	validator UserValidator
+}
+
+func (n *mysqlCachingSha2AuthMethod) Name() AuthMethodDescription {
+	return CachingSha2Password
+}
+
+func (n *mysqlCachingSha2AuthMethod) HandleUser(conn *Conn, user string) bool {
+	if !conn.TLSEnabled() && !conn.IsUnixSocket() {
+		return false
+	}
+	return n.validator.HandleUser(user)
+}
+
+func (n *mysqlCachingSha2AuthMethod) AuthPluginData() ([]byte, error) {
+	salt, err := newSalt()
+	if err != nil {
+		return nil, err
+	}
+	return append(salt, 0), nil
+}
+
+func (n *mysqlCachingSha2AuthMethod) HandleAuthPluginData(c *Conn, user string, serverAuthPluginData []byte, clientAuthPluginData []byte, remoteAddr net.Addr) (Getter, error) {
+	if serverAuthPluginData[len(serverAuthPluginData)-1] != 0x00 {
+		return nil, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+	}
+
+	salt := serverAuthPluginData[:len(serverAuthPluginData)-1]
+	result, cacheState, err := n.cache.UserEntryWithCacheHash(c.GetTLSClientCerts(), salt, user, clientAuthPluginData, remoteAddr)
+
+	if err != nil {
+		return nil, err
+	}
+
+	switch cacheState {
+	case AuthRejected:
+		return nil, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+	case AuthAccepted:
+		// We need to write a more data packet to indicate the
+		// handshake completed properly. This  will be followed
+		// by a regular OK packet which the caller of this method will send.
+		data, pos := c.startEphemeralPacketWithHeader(2)
+		pos = writeByte(data, pos, AuthMoreDataPacket)
+		_ = writeByte(data, pos, CachingSha2FastAuth)
+		err = c.writeEphemeralPacket()
+		if err != nil {
+			return nil, err
+		}
+		return result, nil
+	case AuthNeedMoreData:
+		if !c.TLSEnabled() && !c.IsUnixSocket() {
+			return nil, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+		}
+
+		data, pos := c.startEphemeralPacketWithHeader(1)
+		pos = writeByte(data, pos, AuthMoreDataPacket)
+		writeByte(data, pos, CachingSha2FullAuth)
+		c.writeEphemeralPacket()
+
+		password, err := readPacketPasswordString(c)
+		if err != nil {
+			return nil, err
+		}
+
+		return n.storage.UserEntryWithPassword(c.GetTLSClientCerts(), user, password, remoteAddr)
+	default:
+		// Somehow someone returned an unknown state, let's error with access denied.
+		return nil, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+	}
+}
+
+// authServers is a registry of AuthServer implementations.
+var authServers = make(map[string]AuthServer)
+
+// mu is used to lock access to authServers
+var mu sync.Mutex
+
+// RegisterAuthServer registers an implementations of AuthServer.
+func RegisterAuthServer(name string, authServer AuthServer) {
+	mu.Lock()
+	defer mu.Unlock()
+	if _, ok := authServers[name]; ok {
+		log.Fatalf("AuthServer named %v already exists", name)
+	}
+	authServers[name] = authServer
+}
+
+// GetAuthServer returns an AuthServer by name, or log.Exitf.
+func GetAuthServer(name string) AuthServer {
+	mu.Lock()
+	defer mu.Unlock()
+	authServer, ok := authServers[name]
+	if !ok {
+		log.Exitf("no AuthServer name %v registered", name)
+	}
+	return authServer
+}
+
+func newSalt() ([]byte, error) {
+	salt := make([]byte, 20)
+	if _, err := rand.Read(salt); err != nil {
+		return nil, err
+	}
+
+	// Salt must be a legal UTF8 string.
+	for i := 0; i < len(salt); i++ {
+		salt[i] &= 0x7f
+		if salt[i] == '\x00' || salt[i] == '$' {
+			salt[i]++
+		}
+	}
+
+	return salt, nil
+}
+
+func negotiateAuthMethod(conn *Conn, as AuthServer, user string, requestedAuth AuthMethodDescription) (AuthMethod, error) {
+	for _, m := range as.AuthMethods() {
+		if m.Name() == requestedAuth && m.HandleUser(conn, user) {
+			return m, nil
+		}
+	}
+	return nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "unknown auth method requested: %s", string(requestedAuth))
+}
+
+func readPacketPasswordString(c *Conn) (string, error) {
 	// Read a packet, the password is the payload, as a
 	// zero terminated string.
 	data, err := c.ReadPacket()
@@ -282,20 +616,3 @@ func AuthServerReadPacketString(c *Conn) (string, error) {
 	}
 	return string(data[:len(data)-1]), nil
 }
-
-// AuthServerNegotiateClearOrDialog will finish a negotiation based on
-// the method type for the connection. Only supports
-// MysqlClearPassword and MysqlDialog.
-func AuthServerNegotiateClearOrDialog(c *Conn, method string) (string, error) {
-	switch method {
-	case MysqlClearPassword:
-		// The password is the next packet in plain text.
-		return AuthServerReadPacketString(c)
-
-	case MysqlDialog:
-		return AuthServerReadPacketString(c)
-
-	default:
-		return "", vterrors.Errorf(vtrpc.Code_INTERNAL, "unrecognized method: %v", method)
-	}
-}
diff --git a/go/mysql/auth_server_clientcert.go b/go/mysql/auth_server_clientcert.go
index e860248efb..04f6bb9798 100644
--- a/go/mysql/auth_server_clientcert.go
+++ b/go/mysql/auth_server_clientcert.go
@@ -17,6 +17,7 @@ limitations under the License.
 package mysql
 
 import (
+	"crypto/x509"
 	"flag"
 	"fmt"
 	"net"
@@ -24,58 +25,71 @@ import (
 	"vitess.io/vitess/go/vt/log"
 )
 
-var clientcertAuthMethod = flag.String("mysql_clientcert_auth_method", MysqlClearPassword, "client-side authentication method to use. Supported values: mysql_clear_password, dialog.")
+var clientcertAuthMethod = flag.String("mysql_clientcert_auth_method", string(MysqlClearPassword), "client-side authentication method to use. Supported values: mysql_clear_password, dialog.")
 
+// AuthServerClientCert implements AuthServer which enforces client side certificates
 type AuthServerClientCert struct {
-	Method string
+	methods []AuthMethod
+	Method  AuthMethodDescription
 }
 
-// Init is public so it can be called from plugin_auth_clientcert.go (go/cmd/vtgate)
+// InitAuthServerClientCert is public so it can be called from plugin_auth_clientcert.go (go/cmd/vtgate)
 func InitAuthServerClientCert() {
 	if flag.CommandLine.Lookup("mysql_server_ssl_ca").Value.String() == "" {
 		log.Info("Not configuring AuthServerClientCert because mysql_server_ssl_ca is empty")
 		return
 	}
-	if *clientcertAuthMethod != MysqlClearPassword && *clientcertAuthMethod != MysqlDialog {
+	if *clientcertAuthMethod != string(MysqlClearPassword) && *clientcertAuthMethod != string(MysqlDialog) {
 		log.Exitf("Invalid mysql_clientcert_auth_method value: only support mysql_clear_password or dialog")
 	}
+
+	ascc := newAuthServerClientCert()
+	RegisterAuthServer("clientcert", ascc)
+}
+
+func newAuthServerClientCert() *AuthServerClientCert {
 	ascc := &AuthServerClientCert{
-		Method: *clientcertAuthMethod,
+		Method: AuthMethodDescription(*clientcertAuthMethod),
 	}
-	RegisterAuthServerImpl("clientcert", ascc)
-}
 
-// AuthMethod is part of the AuthServer interface.
-func (ascc *AuthServerClientCert) AuthMethod(user string) (string, error) {
-	return ascc.Method, nil
+	var authMethod AuthMethod
+	switch AuthMethodDescription(*clientcertAuthMethod) {
+	case MysqlClearPassword:
+		authMethod = NewMysqlClearAuthMethod(ascc, ascc)
+	case MysqlDialog:
+		authMethod = NewMysqlDialogAuthMethod(ascc, ascc, "")
+	default:
+		log.Exitf("Invalid mysql_ldap_auth_method value: only support mysql_clear_password or dialog")
+	}
+
+	ascc.methods = []AuthMethod{authMethod}
+	return ascc
 }
 
-// Salt is not used for this plugin.
-func (ascc *AuthServerClientCert) Salt() ([]byte, error) {
-	return NewSalt()
+// AuthMethods returns the implement auth methods for the client
+// certificate authentication setup.
+func (asl *AuthServerClientCert) AuthMethods() []AuthMethod {
+	return asl.methods
 }
 
-// ValidateHash is unimplemented.
-func (ascc *AuthServerClientCert) ValidateHash(salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, error) {
-	panic("unimplemented")
+// DefaultAuthMethodDescription returns always MysqlNativePassword
+// for the client certificate authentication setup.
+func (asl *AuthServerClientCert) DefaultAuthMethodDescription() AuthMethodDescription {
+	return MysqlNativePassword
 }
 
-// Negotiate is part of the AuthServer interface.
-func (ascc *AuthServerClientCert) Negotiate(c *Conn, user string, remoteAddr net.Addr) (Getter, error) {
-	// This code depends on the fact that golang's tls server enforces client cert verification.
-	// Note that the -mysql_server_ssl_ca flag must be set in order for the vtgate to accept client certs.
-	// If not set, the vtgate will effectively deny all incoming mysql connections, since they will all lack certificates.
-	// For more info, check out go/vt/vtttls/vttls.go
-	certs := c.GetTLSClientCerts()
-	if len(certs) == 0 {
-		return nil, fmt.Errorf("no client certs for connection ID %v", c.ConnectionID)
-	}
+// HandleUser is part of the UserValidator interface. We
+// handle any user here since we don't check up front.
+func (asl *AuthServerClientCert) HandleUser(user string) bool {
+	return true
+}
 
-	if _, err := AuthServerReadPacketString(c); err != nil {
-		return nil, err
+// UserEntryWithPassword is part of the PlaintextStorage interface
+func (asl *AuthServerClientCert) UserEntryWithPassword(userCerts []*x509.Certificate, user string, password string, remoteAddr net.Addr) (Getter, error) {
+	if len(userCerts) == 0 {
+		return nil, fmt.Errorf("no client certs for connection")
 	}
-
-	commonName := certs[0].Subject.CommonName
+	commonName := userCerts[0].Subject.CommonName
 
 	if user != commonName {
 		return nil, fmt.Errorf("MySQL connection username '%v' does not match client cert common name '%v'", user, commonName)
@@ -83,6 +97,6 @@ func (ascc *AuthServerClientCert) Negotiate(c *Conn, user string, remoteAddr net
 
 	return &StaticUserData{
 		username: commonName,
-		groups:   certs[0].DNSNames,
+		groups:   userCerts[0].DNSNames,
 	}, nil
 }
diff --git a/go/mysql/auth_server_none.go b/go/mysql/auth_server_none.go
index c560238303..39baa4b357 100644
--- a/go/mysql/auth_server_none.go
+++ b/go/mysql/auth_server_none.go
@@ -17,6 +17,7 @@ limitations under the License.
 package mysql
 
 import (
+	"crypto/x509"
 	"net"
 
 	querypb "vitess.io/vitess/go/vt/proto/query"
@@ -27,32 +28,43 @@ import (
 // With this config, you can connect to a local vtgate using
 // the following command line: 'mysql -P port -h ::'.
 // It only uses MysqlNativePassword method.
-type AuthServerNone struct{}
+type AuthServerNone struct {
+	methods []AuthMethod
+}
 
-// AuthMethod is part of the AuthServer interface.
-// We always return MysqlNativePassword.
-func (a *AuthServerNone) AuthMethod(user string) (string, error) {
-	return MysqlNativePassword, nil
+// AuthMethods returns the list of registered auth methods
+// implemented by this auth server.
+func (a *AuthServerNone) AuthMethods() []AuthMethod {
+	return a.methods
 }
 
-// Salt makes salt
-func (a *AuthServerNone) Salt() ([]byte, error) {
-	return NewSalt()
+// DefaultAuthMethodDescription returns MysqlNativePassword as the default
+// authentication method for the auth server implementation.
+func (a *AuthServerNone) DefaultAuthMethodDescription() AuthMethodDescription {
+	return MysqlNativePassword
 }
 
-// ValidateHash validates hash
-func (a *AuthServerNone) ValidateHash(salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, error) {
-	return &NoneGetter{}, nil
+// HandleUser validates if this user can use this auth method
+func (a *AuthServerNone) HandleUser(user string) bool {
+	return true
 }
 
-// Negotiate is part of the AuthServer interface.
-// It will never be called.
-func (a *AuthServerNone) Negotiate(c *Conn, user string, remotAddr net.Addr) (Getter, error) {
-	panic("Negotiate should not be called as AuthMethod returned mysql_native_password")
+// UserEntryWithHash validates the user if it exists and returns the information.
+// Always accepts any user.
+func (a *AuthServerNone) UserEntryWithHash(userCerts []*x509.Certificate, salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, error) {
+	return &NoneGetter{}, nil
 }
 
 func init() {
-	RegisterAuthServerImpl("none", &AuthServerNone{})
+	a := NewAuthServerNone()
+	RegisterAuthServer("none", a)
+}
+
+// NewAuthServerNone returns an empty auth server. Always accepts all clients.
+func NewAuthServerNone() *AuthServerNone {
+	a := &AuthServerNone{}
+	a.methods = []AuthMethod{NewMysqlNativeAuthMethod(a, a)}
+	return a
 }
 
 // NoneGetter holds the empty string
diff --git a/go/mysql/auth_server_static.go b/go/mysql/auth_server_static.go
index ef95febb29..496cf6c857 100644
--- a/go/mysql/auth_server_static.go
+++ b/go/mysql/auth_server_static.go
@@ -18,6 +18,8 @@ package mysql
 
 import (
 	"bytes"
+	"crypto/subtle"
+	"crypto/x509"
 	"encoding/json"
 	"flag"
 	"io/ioutil"
@@ -46,14 +48,9 @@ const (
 
 // AuthServerStatic implements AuthServer using a static configuration.
 type AuthServerStatic struct {
+	methods          []AuthMethod
 	file, jsonConfig string
 	reloadInterval   time.Duration
-	// method can be set to:
-	// - MysqlNativePassword
-	// - MysqlClearPassword
-	// - MysqlDialog
-	// It defaults to MysqlNativePassword.
-	method string
 	// This mutex helps us prevent data races between the multiple updates of entries.
 	mu sync.Mutex
 	// entries contains the users, passwords and user data.
@@ -109,7 +106,7 @@ func RegisterAuthServerStaticFromParams(file, jsonConfig string, reloadInterval
 	if len(authServerStatic.entries) <= 0 {
 		log.Exitf("Failed to populate entries from file: %v", file)
 	}
-	RegisterAuthServerImpl("static", authServerStatic)
+	RegisterAuthServer("static", authServerStatic)
 }
 
 // NewAuthServerStatic returns a new empty AuthServerStatic.
@@ -118,14 +115,137 @@ func NewAuthServerStatic(file, jsonConfig string, reloadInterval time.Duration)
 		file:           file,
 		jsonConfig:     jsonConfig,
 		reloadInterval: reloadInterval,
-		method:         MysqlNativePassword,
 		entries:        make(map[string][]*AuthServerStaticEntry),
 	}
+
+	a.methods = []AuthMethod{NewMysqlNativeAuthMethod(a, a)}
+
 	a.reload()
 	a.installSignalHandlers()
 	return a
 }
 
+// NewAuthServerStaticWithAuthMethodDescription returns a new empty AuthServerStatic
+// but with support for a different auth method. Mostly used for testing purposes.
+func NewAuthServerStaticWithAuthMethodDescription(file, jsonConfig string, reloadInterval time.Duration, authMethodDescription AuthMethodDescription) *AuthServerStatic {
+	a := &AuthServerStatic{
+		file:           file,
+		jsonConfig:     jsonConfig,
+		reloadInterval: reloadInterval,
+		entries:        make(map[string][]*AuthServerStaticEntry),
+	}
+
+	var authMethod AuthMethod
+	switch authMethodDescription {
+	case CachingSha2Password:
+		authMethod = NewSha2CachingAuthMethod(a, a, a)
+	case MysqlNativePassword:
+		authMethod = NewMysqlNativeAuthMethod(a, a)
+	case MysqlClearPassword:
+		authMethod = NewMysqlClearAuthMethod(a, a)
+	case MysqlDialog:
+		authMethod = NewMysqlDialogAuthMethod(a, a, "")
+	}
+
+	a.methods = []AuthMethod{authMethod}
+
+	a.reload()
+	a.installSignalHandlers()
+	return a
+}
+
+// HandleUser is part of the Validator interface. We
+// handle any user here since we don't check up front.
+func (a *AuthServerStatic) HandleUser(user string) bool {
+	return true
+}
+
+// UserEntryWithPassword implements password lookup based on a plain
+// text password that is negotiated with the client.
+func (a *AuthServerStatic) UserEntryWithPassword(userCerts []*x509.Certificate, user string, password string, remoteAddr net.Addr) (Getter, error) {
+	a.mu.Lock()
+	entries, ok := a.entries[user]
+	a.mu.Unlock()
+
+	if !ok {
+		return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+	}
+
+	for _, entry := range entries {
+		// Validate the password.
+		if matchSourceHost(remoteAddr, entry.SourceHost) && subtle.ConstantTimeCompare([]byte(password), []byte(entry.Password)) == 1 {
+			return &StaticUserData{entry.UserData, entry.Groups}, nil
+		}
+	}
+	return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+}
+
+// UserEntryWithHash implements password lookup based on a
+// mysql_native_password hash that is negotiated with the client.
+func (a *AuthServerStatic) UserEntryWithHash(userCerts []*x509.Certificate, salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, error) {
+	a.mu.Lock()
+	entries, ok := a.entries[user]
+	a.mu.Unlock()
+
+	if !ok {
+		return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+	}
+
+	for _, entry := range entries {
+		if entry.MysqlNativePassword != "" {
+			hash, err := DecodeMysqlNativePasswordHex(entry.MysqlNativePassword)
+			if err != nil {
+				return &StaticUserData{entry.UserData, entry.Groups}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+			}
+
+			isPass := VerifyHashedMysqlNativePassword(authResponse, salt, hash)
+			if matchSourceHost(remoteAddr, entry.SourceHost) && isPass {
+				return &StaticUserData{entry.UserData, entry.Groups}, nil
+			}
+		} else {
+			computedAuthResponse := ScrambleMysqlNativePassword(salt, []byte(entry.Password))
+			// Validate the password.
+			if matchSourceHost(remoteAddr, entry.SourceHost) && subtle.ConstantTimeCompare(authResponse, computedAuthResponse) == 1 {
+				return &StaticUserData{entry.UserData, entry.Groups}, nil
+			}
+		}
+	}
+	return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+}
+
+// UserEntryWithCacheHash implements password lookup based on a
+// caching_sha2_password hash that is negotiated with the client.
+func (a *AuthServerStatic) UserEntryWithCacheHash(userCerts []*x509.Certificate, salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, CacheState, error) {
+	a.mu.Lock()
+	entries, ok := a.entries[user]
+	a.mu.Unlock()
+
+	if !ok {
+		return &StaticUserData{}, AuthRejected, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+	}
+
+	for _, entry := range entries {
+		computedAuthResponse := ScrambleCachingSha2Password(salt, []byte(entry.Password))
+
+		// Validate the password.
+		if matchSourceHost(remoteAddr, entry.SourceHost) && subtle.ConstantTimeCompare(authResponse, computedAuthResponse) == 1 {
+			return &StaticUserData{entry.UserData, entry.Groups}, AuthAccepted, nil
+		}
+	}
+	return &StaticUserData{}, AuthRejected, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+}
+
+// AuthMethods returns the AuthMethod instances this auth server can handle.
+func (a *AuthServerStatic) AuthMethods() []AuthMethod {
+	return a.methods
+}
+
+// DefaultAuthMethodDescription returns the default auth method in the handshake which
+// is MysqlNativePassword for this auth server.
+func (a *AuthServerStatic) DefaultAuthMethodDescription() AuthMethodDescription {
+	return MysqlNativePassword
+}
+
 func (a *AuthServerStatic) reload() {
 	jsonBytes := []byte(a.jsonConfig)
 	if a.file != "" {
@@ -217,69 +337,6 @@ func validateConfig(config map[string][]*AuthServerStaticEntry) error {
 	return nil
 }
 
-// AuthMethod is part of the AuthServer interface.
-func (a *AuthServerStatic) AuthMethod(user string) (string, error) {
-	return a.method, nil
-}
-
-// Salt is part of the AuthServer interface.
-func (a *AuthServerStatic) Salt() ([]byte, error) {
-	return NewSalt()
-}
-
-// ValidateHash is part of the AuthServer interface.
-func (a *AuthServerStatic) ValidateHash(salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, error) {
-	a.mu.Lock()
-	entries, ok := a.entries[user]
-	a.mu.Unlock()
-
-	if !ok {
-		return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
-	}
-
-	for _, entry := range entries {
-		if entry.MysqlNativePassword != "" {
-			isPass := isPassScrambleMysqlNativePassword(authResponse, salt, entry.MysqlNativePassword)
-			if matchSourceHost(remoteAddr, entry.SourceHost) && isPass {
-				return &StaticUserData{entry.UserData, entry.Groups}, nil
-			}
-		} else {
-			computedAuthResponse := ScrambleMysqlNativePassword(salt, []byte(entry.Password))
-			// Validate the password.
-			if matchSourceHost(remoteAddr, entry.SourceHost) && bytes.Equal(authResponse, computedAuthResponse) {
-				return &StaticUserData{entry.UserData, entry.Groups}, nil
-			}
-		}
-	}
-	return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
-}
-
-// Negotiate is part of the AuthServer interface.
-// It will be called if method is anything else than MysqlNativePassword.
-// We only recognize MysqlClearPassword and MysqlDialog here.
-func (a *AuthServerStatic) Negotiate(c *Conn, user string, remoteAddr net.Addr) (Getter, error) {
-	// Finish the negotiation.
-	password, err := AuthServerNegotiateClearOrDialog(c, a.method)
-	if err != nil {
-		return nil, err
-	}
-
-	a.mu.Lock()
-	entries, ok := a.entries[user]
-	a.mu.Unlock()
-
-	if !ok {
-		return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
-	}
-	for _, entry := range entries {
-		// Validate the password.
-		if matchSourceHost(remoteAddr, entry.SourceHost) && entry.Password == password {
-			return &StaticUserData{entry.UserData, entry.Groups}, nil
-		}
-	}
-	return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
-}
-
 func matchSourceHost(remoteAddr net.Addr, targetSourceHost string) bool {
 	// Legacy support, there was not matcher defined default to true
 	if targetSourceHost == "" {
diff --git a/go/mysql/auth_server_vault.go b/go/mysql/auth_server_vault.go
index 25fe6611a4..b0a8c1fe49 100644
--- a/go/mysql/auth_server_vault.go
+++ b/go/mysql/auth_server_vault.go
@@ -17,7 +17,8 @@ limitations under the License.
 package mysql
 
 import (
-	"bytes"
+	"crypto/subtle"
+	"crypto/x509"
 	"flag"
 	"fmt"
 	"io/ioutil"
@@ -48,16 +49,11 @@ var (
 
 // AuthServerVault implements AuthServer with a config loaded from Vault.
 type AuthServerVault struct {
-	mu sync.Mutex
-	// method can be set to:
-	// - MysqlNativePassword
-	// - MysqlClearPassword
-	// - MysqlDialog
-	// It defaults to MysqlNativePassword.
-	method string
+	methods []AuthMethod
+	mu      sync.Mutex
 	// users, passwords and user data
-	//   We use the same JSON format as for -mysql_auth_server_static
-	//   Acts as a cache for the in-Vault data
+	// We use the same JSON format as for -mysql_auth_server_static
+	// Acts as a cache for the in-Vault data
 	entries                map[string][]*AuthServerStaticEntry
 	vaultCacheExpireTicker *time.Ticker
 	vaultClient            *vaultapi.Client
@@ -86,7 +82,7 @@ func registerAuthServerVault(addr string, timeout time.Duration, caCertPath stri
 	if err != nil {
 		log.Exitf("%s", err)
 	}
-	RegisterAuthServerImpl("vault", authServerVault)
+	RegisterAuthServer("vault", authServerVault)
 }
 
 func newAuthServerVault(addr string, timeout time.Duration, caCertPath string, path string, ttl time.Duration, tokenFilePath string, roleID string, secretIDPath string, roleMountPoint string) (*AuthServerVault, error) {
@@ -140,15 +136,66 @@ func newAuthServerVault(addr string, timeout time.Duration, caCertPath string, p
 		vaultClient: client,
 		vaultPath:   path,
 		vaultTTL:    ttl,
-		method:      MysqlNativePassword,
 		entries:     make(map[string][]*AuthServerStaticEntry),
 	}
 
+	authMethodNative := NewMysqlNativeAuthMethod(a, a)
+	a.methods = []AuthMethod{authMethodNative}
+
 	a.reloadVault()
 	a.installSignalHandlers()
 	return a, nil
 }
 
+// AuthMethods returns the list of registered auth methods
+// implemented by this auth server.
+func (a *AuthServerVault) AuthMethods() []AuthMethod {
+	return a.methods
+}
+
+// DefaultAuthMethodDescription returns MysqlNativePassword as the default
+// authentication method for the auth server implementation.
+func (a *AuthServerVault) DefaultAuthMethodDescription() AuthMethodDescription {
+	return MysqlNativePassword
+}
+
+// HandleUser is part of the Validator interface. We
+// handle any user here since we don't check up front.
+func (a *AuthServerVault) HandleUser(user string) bool {
+	return true
+}
+
+// UserEntryWithHash is called when mysql_native_password is used.
+func (a *AuthServerVault) UserEntryWithHash(userCerts []*x509.Certificate, salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, error) {
+	a.mu.Lock()
+	userEntries, ok := a.entries[user]
+	a.mu.Unlock()
+
+	if !ok {
+		return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+	}
+
+	for _, entry := range userEntries {
+		if entry.MysqlNativePassword != "" {
+			hash, err := DecodeMysqlNativePasswordHex(entry.MysqlNativePassword)
+			if err != nil {
+				return &StaticUserData{entry.UserData, entry.Groups}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+			}
+			isPass := VerifyHashedMysqlNativePassword(authResponse, salt, hash)
+			if matchSourceHost(remoteAddr, entry.SourceHost) && isPass {
+				return &StaticUserData{entry.UserData, entry.Groups}, nil
+			}
+		} else {
+			computedAuthResponse := ScrambleMysqlNativePassword(salt, []byte(entry.Password))
+			// Validate the password.
+			if matchSourceHost(remoteAddr, entry.SourceHost) && subtle.ConstantTimeCompare(authResponse, computedAuthResponse) == 1 {
+				return &StaticUserData{entry.UserData, entry.Groups}, nil
+			}
+		}
+	}
+	return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
+}
+
 func (a *AuthServerVault) setTTLTicker(ttl time.Duration) {
 	a.mu.Lock()
 	defer a.mu.Unlock()
@@ -224,69 +271,6 @@ func (a *AuthServerVault) close() {
 	}
 }
 
-// AuthMethod is part of the AuthServer interface.
-func (a *AuthServerVault) AuthMethod(user string) (string, error) {
-	return a.method, nil
-}
-
-// Salt is part of the AuthServer interface.
-func (a *AuthServerVault) Salt() ([]byte, error) {
-	return NewSalt()
-}
-
-// ValidateHash is part of the AuthServer interface.
-func (a *AuthServerVault) ValidateHash(salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, error) {
-	a.mu.Lock()
-	userEntries, ok := a.entries[user]
-	a.mu.Unlock()
-
-	if !ok {
-		return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
-	}
-
-	for _, entry := range userEntries {
-		if entry.MysqlNativePassword != "" {
-			isPass := isPassScrambleMysqlNativePassword(authResponse, salt, entry.MysqlNativePassword)
-			if matchSourceHost(remoteAddr, entry.SourceHost) && isPass {
-				return &StaticUserData{entry.UserData, entry.Groups}, nil
-			}
-		} else {
-			computedAuthResponse := ScrambleMysqlNativePassword(salt, []byte(entry.Password))
-			// Validate the password.
-			if matchSourceHost(remoteAddr, entry.SourceHost) && bytes.Equal(authResponse, computedAuthResponse) {
-				return &StaticUserData{entry.UserData, entry.Groups}, nil
-			}
-		}
-	}
-	return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
-}
-
-// Negotiate is part of the AuthServer interface.
-// It will be called if method is anything else than MysqlNativePassword.
-// We only recognize MysqlClearPassword and MysqlDialog here.
-func (a *AuthServerVault) Negotiate(c *Conn, user string, remoteAddr net.Addr) (Getter, error) {
-	// Finish the negotiation.
-	password, err := AuthServerNegotiateClearOrDialog(c, a.method)
-	if err != nil {
-		return nil, err
-	}
-
-	a.mu.Lock()
-	userEntries, ok := a.entries[user]
-	a.mu.Unlock()
-
-	if !ok {
-		return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
-	}
-	for _, entry := range userEntries {
-		// Validate the password.
-		if matchSourceHost(remoteAddr, entry.SourceHost) && entry.Password == password {
-			return &StaticUserData{entry.UserData, entry.Groups}, nil
-		}
-	}
-	return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
-}
-
 // We ignore most errors here, to allow us to retry cleanly
 //   or ignore the cases where the input is not passed by file, but via env
 func readFromFile(filePath string) (string, error) {
diff --git a/go/mysql/client.go b/go/mysql/client.go
index b63ce69205..dee5d8e483 100644
--- a/go/mysql/client.go
+++ b/go/mysql/client.go
@@ -483,7 +483,7 @@ func (c *Conn) parseInitialHandshakePacket(data []byte) (uint32, []byte, error)
 			// 5.6.2 that don't have a null terminated string.
 			authPluginName = string(data[pos : len(data)-1])
 		}
-		c.authPluginName = authPluginName
+		c.authPluginName = AuthMethodDescription(authPluginName)
 	}
 
 	return capabilities, authPluginData, nil
@@ -617,7 +617,7 @@ func (c *Conn) writeHandshakeResponse41(capabilities uint32, scrambledPassword [
 	}
 
 	// Assume native client during response
-	pos = writeNullString(data, pos, c.authPluginName)
+	pos = writeNullString(data, pos, string(c.authPluginName))
 
 	// Sanity-check the length.
 	if pos != len(data) {
@@ -734,7 +734,7 @@ func (c *Conn) handleAuthMoreDataPacket(data byte, params *ConnParams) error {
 	}
 }
 
-func parseAuthSwitchRequest(data []byte) (string, []byte, error) {
+func parseAuthSwitchRequest(data []byte) (AuthMethodDescription, []byte, error) {
 	pos := 1
 	pluginName, pos, ok := readNullString(data, pos)
 	if !ok {
@@ -746,7 +746,7 @@ func parseAuthSwitchRequest(data []byte) (string, []byte, error) {
 	if len(salt) > 20 {
 		salt = salt[:20]
 	}
-	return pluginName, salt, nil
+	return AuthMethodDescription(pluginName), salt, nil
 }
 
 // requestPublicKey requests a public key from the server
diff --git a/go/mysql/conn.go b/go/mysql/conn.go
index 834cf60afc..0c557d143b 100644
--- a/go/mysql/conn.go
+++ b/go/mysql/conn.go
@@ -91,7 +91,7 @@ type Conn struct {
 
 	// authPluginName is the name of server's authentication plugin.
 	// It is set during the initial handshake.
-	authPluginName string
+	authPluginName AuthMethodDescription
 
 	// schemaName is the default database name to use. It is set
 	// during handshake, and by ComInitDb packets. Both client and
@@ -1499,6 +1499,17 @@ func (c *Conn) GetTLSClientCerts() []*x509.Certificate {
 	return nil
 }
 
+// TLSEnabled returns true if this connection is using TLS.
+func (c *Conn) TLSEnabled() bool {
+	return c.Capabilities&CapabilityClientSSL > 0
+}
+
+// IsUnixSocket returns true if this connection is over a Unix socket.
+func (c *Conn) IsUnixSocket() bool {
+	_, ok := c.listener.listener.(*net.UnixListener)
+	return ok
+}
+
 // GetRawConn returns the raw net.Conn for nefarious purposes.
 func (c *Conn) GetRawConn() net.Conn {
 	return c.conn
diff --git a/go/mysql/constants.go b/go/mysql/constants.go
index cd18e49b26..99cf5ff627 100644
--- a/go/mysql/constants.go
+++ b/go/mysql/constants.go
@@ -34,20 +34,24 @@ const (
 	protocolVersion = 10
 )
 
+// AuthMethodDescription is the type for different supported and
+// implemented authentication methods.
+type AuthMethodDescription string
+
 // Supported auth forms.
 const (
 	// MysqlNativePassword uses a salt and transmits a hash on the wire.
-	MysqlNativePassword = "mysql_native_password"
+	MysqlNativePassword = AuthMethodDescription("mysql_native_password")
 
 	// MysqlClearPassword transmits the password in the clear.
-	MysqlClearPassword = "mysql_clear_password"
+	MysqlClearPassword = AuthMethodDescription("mysql_clear_password")
 
 	// CachingSha2Password uses a salt and transmits a SHA256 hash on the wire.
-	CachingSha2Password = "caching_sha2_password"
+	CachingSha2Password = AuthMethodDescription("caching_sha2_password")
 
 	// MysqlDialog uses the dialog plugin on the client side.
 	// It transmits data in the clear.
-	MysqlDialog = "dialog"
+	MysqlDialog = AuthMethodDescription("dialog")
 )
 
 // Capability flags.
diff --git a/go/mysql/fakesqldb/server.go b/go/mysql/fakesqldb/server.go
index 863ee2bbd9..065c3400ec 100644
--- a/go/mysql/fakesqldb/server.go
+++ b/go/mysql/fakesqldb/server.go
@@ -173,7 +173,7 @@ func New(t testing.TB) *DB {
 
 	db.Handler = db
 
-	authServer := &mysql.AuthServerNone{}
+	authServer := mysql.NewAuthServerNone()
 
 	// Start listening.
 	db.listener, err = mysql.NewListener("unix", socketFile, authServer, db, 0, 0, false)
diff --git a/go/mysql/ldapauthserver/auth_server_ldap.go b/go/mysql/ldapauthserver/auth_server_ldap.go
index 172b20335c..727b28b5b1 100644
--- a/go/mysql/ldapauthserver/auth_server_ldap.go
+++ b/go/mysql/ldapauthserver/auth_server_ldap.go
@@ -17,6 +17,7 @@ limitations under the License.
 package ldapauthserver
 
 import (
+	"crypto/x509"
 	"encoding/json"
 	"flag"
 	"fmt"
@@ -37,19 +38,19 @@ import (
 var (
 	ldapAuthConfigFile   = flag.String("mysql_ldap_auth_config_file", "", "JSON File from which to read LDAP server config.")
 	ldapAuthConfigString = flag.String("mysql_ldap_auth_config_string", "", "JSON representation of LDAP server config.")
-	ldapAuthMethod       = flag.String("mysql_ldap_auth_method", mysql.MysqlClearPassword, "client-side authentication method to use. Supported values: mysql_clear_password, dialog.")
+	ldapAuthMethod       = flag.String("mysql_ldap_auth_method", string(mysql.MysqlClearPassword), "client-side authentication method to use. Supported values: mysql_clear_password, dialog.")
 )
 
 // AuthServerLdap implements AuthServer with an LDAP backend
 type AuthServerLdap struct {
 	Client
 	ServerConfig
-	Method         string
 	User           string
 	Password       string
 	GroupQuery     string
 	UserDnPattern  string
 	RefreshSeconds int64
+	methods        []mysql.AuthMethod
 }
 
 // Init is public so it can be called from plugin_auth_ldap.go (go/cmd/vtgate)
@@ -62,13 +63,13 @@ func Init() {
 		log.Infof("Both mysql_ldap_auth_config_file and mysql_ldap_auth_config_string are non-empty, can only use one.")
 		return
 	}
-	if *ldapAuthMethod != mysql.MysqlClearPassword && *ldapAuthMethod != mysql.MysqlDialog {
+
+	if *ldapAuthMethod != string(mysql.MysqlClearPassword) && *ldapAuthMethod != string(mysql.MysqlDialog) {
 		log.Exitf("Invalid mysql_ldap_auth_method value: only support mysql_clear_password or dialog")
 	}
 	ldapAuthServer := &AuthServerLdap{
 		Client:       &ClientImpl{},
 		ServerConfig: ServerConfig{},
-		Method:       *ldapAuthMethod,
 	}
 
 	data := []byte(*ldapAuthConfigString)
@@ -82,31 +83,42 @@ func Init() {
 	if err := json.Unmarshal(data, ldapAuthServer); err != nil {
 		log.Exitf("Error parsing AuthServerLdap config: %v", err)
 	}
-	mysql.RegisterAuthServerImpl("ldap", ldapAuthServer)
+
+	var authMethod mysql.AuthMethod
+	switch mysql.AuthMethodDescription(*ldapAuthMethod) {
+	case mysql.MysqlClearPassword:
+		authMethod = mysql.NewMysqlClearAuthMethod(ldapAuthServer, ldapAuthServer)
+	case mysql.MysqlDialog:
+		authMethod = mysql.NewMysqlDialogAuthMethod(ldapAuthServer, ldapAuthServer, "")
+	default:
+		log.Exitf("Invalid mysql_ldap_auth_method value: only support mysql_clear_password or dialog")
+	}
+
+	ldapAuthServer.methods = []mysql.AuthMethod{authMethod}
+	mysql.RegisterAuthServer("ldap", ldapAuthServer)
 }
 
-// AuthMethod is part of the AuthServer interface.
-func (asl *AuthServerLdap) AuthMethod(user string) (string, error) {
-	return asl.Method, nil
+// AuthMethods returns the list of registered auth methods
+// implemented by this auth server.
+func (asl *AuthServerLdap) AuthMethods() []mysql.AuthMethod {
+	return asl.methods
 }
 
-// Salt will be unused in AuthServerLdap.
-func (asl *AuthServerLdap) Salt() ([]byte, error) {
-	return mysql.NewSalt()
+// DefaultAuthMethodDescription returns MysqlNativePassword as the default
+// authentication method for the auth server implementation.
+func (asl *AuthServerLdap) DefaultAuthMethodDescription() mysql.AuthMethodDescription {
+	return mysql.MysqlNativePassword
 }
 
-// ValidateHash is unimplemented for AuthServerLdap.
-func (asl *AuthServerLdap) ValidateHash(salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (mysql.Getter, error) {
-	panic("unimplemented")
+// HandleUser is part of the Validator interface. We
+// handle any user here since we don't check up front.
+func (asl *AuthServerLdap) HandleUser(user string) bool {
+	return true
 }
 
-// Negotiate is part of the AuthServer interface.
-func (asl *AuthServerLdap) Negotiate(c *mysql.Conn, user string, remoteAddr net.Addr) (mysql.Getter, error) {
-	// Finish the negotiation.
-	password, err := mysql.AuthServerNegotiateClearOrDialog(c, asl.Method)
-	if err != nil {
-		return nil, err
-	}
+// UserEntryWithPassword is part of the PlaintextStorage interface
+// and called after the password is sent by the client.
+func (asl *AuthServerLdap) UserEntryWithPassword(userCerts []*x509.Certificate, user string, password string, remoteAddr net.Addr) (mysql.Getter, error) {
 	return asl.validate(user, password)
 }
 
diff --git a/go/mysql/server.go b/go/mysql/server.go
index 245c9a8bf6..7397d2deeb 100644
--- a/go/mysql/server.go
+++ b/go/mysql/server.go
@@ -317,7 +317,7 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 	defer connCount.Add(-1)
 
 	// First build and send the server handshake packet.
-	salt, err := c.writeHandshakeV10(l.ServerVersion, l.authServer, l.TLSConfig.Load() != nil)
+	serverAuthPluginData, err := c.writeHandshakeV10(l.ServerVersion, l.authServer, l.TLSConfig.Load() != nil)
 	if err != nil {
 		if err != io.EOF {
 			log.Errorf("Cannot send HandshakeV10 packet to %s: %v", c, err)
@@ -335,7 +335,7 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 		}
 		return
 	}
-	user, authMethod, authResponse, err := l.parseClientHandshakePacket(c, true, response)
+	user, clientAuthMethod, clientAuthResponse, err := l.parseClientHandshakePacket(c, true, response)
 	if err != nil {
 		log.Errorf("Cannot parse client handshake response from %s: %v", c, err)
 		return
@@ -343,7 +343,7 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 
 	c.recycleReadPacket()
 
-	if c.Capabilities&CapabilityClientSSL > 0 {
+	if c.TLSEnabled() {
 		// SSL was enabled. We need to re-read the auth packet.
 		response, err = c.readEphemeralPacket()
 		if err != nil {
@@ -352,7 +352,7 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 		}
 
 		// Returns copies of the data, so we can recycle the buffer.
-		user, authMethod, authResponse, err = l.parseClientHandshakePacket(c, false, response)
+		user, clientAuthMethod, clientAuthResponse, err = l.parseClientHandshakePacket(c, false, response)
 		if err != nil {
 			log.Errorf("Cannot parse post-SSL client handshake response from %s: %v", c, err)
 			return
@@ -370,96 +370,65 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 	} else {
 		if l.RequireSecureTransport {
 			c.writeErrorPacketFromError(vterrors.Errorf(vtrpc.Code_UNAVAILABLE, "server does not allow insecure connections, client must use SSL/TLS"))
-			return
 		}
 		connCountByTLSVer.Add(versionNoTLS, 1)
 		defer connCountByTLSVer.Add(versionNoTLS, -1)
 	}
 
 	// See what auth method the AuthServer wants to use for that user.
-	authServerMethod, err := l.authServer.AuthMethod(user)
+	negotiatedAuthMethod, err := negotiateAuthMethod(c, l.authServer, user, clientAuthMethod)
 	if err != nil {
-		c.writeErrorPacketFromError(err)
-		return
-	}
-
-	// Compare with what the client sent back.
-	switch {
-	case authServerMethod == MysqlNativePassword && authMethod == MysqlNativePassword:
-		// Both server and client want to use MysqlNativePassword:
-		// the negotiation can be completed right away, using the
-		// ValidateHash() method.
-		userData, err := l.authServer.ValidateHash(salt, user, authResponse, conn.RemoteAddr())
-		if err != nil {
-			log.Warningf("Error authenticating user using MySQL native password: %v", err)
-			c.writeErrorPacketFromError(err)
-			return
+		// The client will disconnect if it doesn't understand
+		// the first auth method that we send, so we only have to send the
+		// first one that we allow for the user.
+		for _, m := range l.authServer.AuthMethods() {
+			if m.HandleUser(c, user) {
+				negotiatedAuthMethod = m
+				break
+			}
 		}
-		c.User = user
-		c.UserData = userData
 
-	case authServerMethod == MysqlNativePassword:
-		// The server really wants to use MysqlNativePassword,
-		// but the client returned a result for something else.
-
-		salt, err := l.authServer.Salt()
-		if err != nil {
-			return
-		}
-		// The binary protocol requires padding with 0
-		data := append(salt, byte(0x00))
-		if err := c.writeAuthSwitchRequest(MysqlNativePassword, data); err != nil {
-			log.Errorf("Error writing auth switch packet for %s: %v", c, err)
+		if negotiatedAuthMethod == nil {
+			c.writeErrorPacket(CRServerHandshakeErr, SSUnknownSQLState, "No authentication methods available for authentication.")
 			return
 		}
 
-		response, err := c.readEphemeralPacket()
-		if err != nil {
-			log.Errorf("Error reading auth switch response for %s: %v", c, err)
-			return
+		if negotiatedAuthMethod.Name() == MysqlClearPassword || negotiatedAuthMethod.Name() == MysqlDialog {
+			if !l.AllowClearTextWithoutTLS.Get() && !c.TLSEnabled() {
+				c.writeErrorPacket(CRServerHandshakeErr, SSUnknownSQLState, "Cannot use clear text authentication over non-SSL connections.")
+				return
+			}
 		}
-		c.recycleReadPacket()
 
-		userData, err := l.authServer.ValidateHash(salt, user, response, conn.RemoteAddr())
+		serverAuthPluginData, err = negotiatedAuthMethod.AuthPluginData()
 		if err != nil {
-			log.Warningf("Error authenticating user using MySQL native password: %v", err)
-			c.writeErrorPacketFromError(err)
+			log.Errorf("Error generating auth switch packet for %s: %v", c, err)
 			return
 		}
-		c.User = user
-		c.UserData = userData
-
-	default:
-		// The server wants to use something else, re-negotiate.
 
-		// The negotiation happens in clear text. Let's check we can.
-		if !l.AllowClearTextWithoutTLS.Get() && c.Capabilities&CapabilityClientSSL == 0 {
-			c.writeErrorPacket(CRServerHandshakeErr, SSUnknownSQLState, "Cannot use clear text authentication over non-SSL connections.")
-			return
-		}
-
-		// Switch our auth method to what the server wants.
-		// Dialog plugin expects an AskPassword prompt.
-		var data []byte
-		if authServerMethod == MysqlDialog {
-			data = authServerDialogSwitchData()
-		}
-		if err := c.writeAuthSwitchRequest(authServerMethod, data); err != nil {
+		if err := c.writeAuthSwitchRequest(string(negotiatedAuthMethod.Name()), serverAuthPluginData); err != nil {
 			log.Errorf("Error writing auth switch packet for %s: %v", c, err)
 			return
 		}
 
-		// Then hand over the rest of the negotiation to the
-		// auth server.
-		userData, err := l.authServer.Negotiate(c, user, conn.RemoteAddr())
+		clientAuthResponse, err = c.readEphemeralPacket()
 		if err != nil {
-			c.writeErrorPacketFromError(err)
+			log.Errorf("Error reading auth switch response for %s: %v", c, err)
 			return
 		}
-		c.User = user
-		c.UserData = userData
+		c.recycleReadPacket()
 	}
 
+	userData, err := negotiatedAuthMethod.HandleAuthPluginData(c, user, serverAuthPluginData, clientAuthResponse, conn.RemoteAddr())
+	if err != nil {
+		log.Warningf("Error authenticating user %s using: %s", user, negotiatedAuthMethod.Name())
+		c.writeErrorPacketFromError(err)
+		return
+	}
+
+	c.User = user
+	c.UserData = userData
+
 	if c.User != "" {
 		connCountPerUser.Add(c.User, 1)
 		defer connCountPerUser.Add(c.User, -1)
@@ -537,11 +506,23 @@ func (c *Conn) writeHandshakeV10(serverVersion string, authServer AuthServer, en
 		capabilities |= CapabilityClientSSL
 	}
 
+	// Grab the default auth method. This can only be either
+	// mysql_native_password or caching_sha2_password. Both
+	// need the salt as well to be present too.
+	//
+	// Any other auth method will cause clients to throw a
+	// handshake error.
+	authMethod := authServer.DefaultAuthMethodDescription()
+
+	if authMethod != MysqlNativePassword && authMethod != CachingSha2Password {
+		authMethod = MysqlNativePassword
+	}
+
 	length :=
 		1 + // protocol version
 			lenNullString(serverVersion) +
 			4 + // connection ID
-			8 + // first part of salt data
+			8 + // first part of plugin auth data
 			1 + // filler byte
 			2 + // capability flags (lower 2 bytes)
 			1 + // character set
@@ -550,7 +531,7 @@ func (c *Conn) writeHandshakeV10(serverVersion string, authServer AuthServer, en
 			1 + // length of auth plugin data
 			10 + // reserved (0)
 			13 + // auth-plugin-data
-			lenNullString(MysqlNativePassword) // auth-plugin-name
+			lenNullString(string(authMethod)) // auth-plugin-name
 
 	data, pos := c.startEphemeralPacketWithHeader(length)
 
@@ -563,13 +544,17 @@ func (c *Conn) writeHandshakeV10(serverVersion string, authServer AuthServer, en
 	// Add connectionID in.
 	pos = writeUint32(data, pos, c.ConnectionID)
 
-	// Generate the salt, put 8 bytes in.
-	salt, err := authServer.Salt()
+	// Generate the salt as the plugin data. Will be reused
+	// later on if no auth method switch happens and the real
+	// auth method is also mysql_native_password or caching_sha2_password.
+	pluginData, err := newSalt()
 	if err != nil {
 		return nil, err
 	}
+	// Plugin data is always defined as having a trailing NULL
+	pluginData = append(pluginData, 0)
 
-	pos += copy(data[pos:], salt[:8])
+	pos += copy(data[pos:], pluginData[:8])
 
 	// One filler byte, always 0.
 	pos = writeByte(data, pos, 0)
@@ -594,12 +579,11 @@ func (c *Conn) writeHandshakeV10(serverVersion string, authServer AuthServer, en
 	pos = writeZeroes(data, pos, 10)
 
 	// Second part of auth plugin data.
-	pos += copy(data[pos:], salt[8:])
-	data[pos] = 0
-	pos++
+	pos += copy(data[pos:], pluginData[8:])
 
-	// Copy authPluginName. We always start with mysql_native_password.
-	pos = writeNullString(data, pos, MysqlNativePassword)
+	// Copy authPluginName. We always start with the first
+	// registered auth method name.
+	pos = writeNullString(data, pos, string(authMethod))
 
 	// Sanity check.
 	if pos != len(data) {
@@ -616,13 +600,13 @@ func (c *Conn) writeHandshakeV10(serverVersion string, authServer AuthServer, en
 		return nil, err
 	}
 
-	return salt, nil
+	return pluginData, nil
 }
 
 // parseClientHandshakePacket parses the handshake sent by the client.
 // Returns the username, auth method, auth data, error.
 // The original data is not pointed at, and can be freed.
-func (l *Listener) parseClientHandshakePacket(c *Conn, firstTime bool, data []byte) (string, string, []byte, error) {
+func (l *Listener) parseClientHandshakePacket(c *Conn, firstTime bool, data []byte) (string, AuthMethodDescription, []byte, error) {
 	pos := 0
 
 	// Client flags, 4 bytes.
@@ -725,15 +709,15 @@ func (l *Listener) parseClientHandshakePacket(c *Conn, firstTime bool, data []by
 	// authMethod (with default)
 	authMethod := MysqlNativePassword
 	if clientFlags&CapabilityClientPluginAuth != 0 {
-		authMethod, pos, ok = readNullString(data, pos)
+		var authMethodStr string
+		authMethodStr, pos, ok = readNullString(data, pos)
 		if !ok {
 			return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read authMethod")
 		}
-	}
-
-	// The JDBC driver sometimes sends an empty string as the auth method when it wants to use mysql_native_password
-	if authMethod == "" {
-		authMethod = MysqlNativePassword
+		// The JDBC driver sometimes sends an empty string as the auth method when it wants to use mysql_native_password
+		if authMethodStr != "" {
+			authMethod = AuthMethodDescription(authMethodStr)
+		}
 	}
 
 	// Decode connection attributes send by the client
@@ -743,7 +727,7 @@ func (l *Listener) parseClientHandshakePacket(c *Conn, firstTime bool, data []by
 		}
 	}
 
-	return username, authMethod, authResponse, nil
+	return username, AuthMethodDescription(authMethod), authResponse, nil
 }
 
 func parseConnAttrs(data []byte, pos int) (map[string]string, int, error) {
diff --git a/go/vt/sqlparser/ast_format.go b/go/vt/sqlparser/ast_format.go
index ad0c45f8fd..0b01a9971a 100644
--- a/go/vt/sqlparser/ast_format.go
+++ b/go/vt/sqlparser/ast_format.go
@@ -473,9 +473,8 @@ func (ct *ColumnType) Format(buf *TrackedBuffer) {
 	if ct.Options.Default != nil {
 		buf.astPrintf(ct, " %s", keywordStrings[DEFAULT])
 		_, isLiteral := ct.Options.Default.(*Literal)
-		_, isBool := ct.Options.Default.(BoolVal)
 		_, isNullVal := ct.Options.Default.(*NullVal)
-		if isLiteral || isNullVal || isBool || isExprAliasForCurrentTimeStamp(ct.Options.Default) {
+		if isLiteral || isNullVal || isExprAliasForCurrentTimeStamp(ct.Options.Default) {
 			buf.astPrintf(ct, " %v", ct.Options.Default)
 		} else {
 			buf.astPrintf(ct, " (%v)", ct.Options.Default)
diff --git a/go/vt/sqlparser/ast_format_fast.go b/go/vt/sqlparser/ast_format_fast.go
index 3f96cd647f..05887bd51e 100644
--- a/go/vt/sqlparser/ast_format_fast.go
+++ b/go/vt/sqlparser/ast_format_fast.go
@@ -673,9 +673,8 @@ func (ct *ColumnType) formatFast(buf *TrackedBuffer) {
 		buf.WriteByte(' ')
 		buf.WriteString(keywordStrings[DEFAULT])
 		_, isLiteral := ct.Options.Default.(*Literal)
-		_, isBool := ct.Options.Default.(BoolVal)
 		_, isNullVal := ct.Options.Default.(*NullVal)
-		if isLiteral || isNullVal || isBool || isExprAliasForCurrentTimeStamp(ct.Options.Default) {
+		if isLiteral || isNullVal || isExprAliasForCurrentTimeStamp(ct.Options.Default) {
 			buf.WriteByte(' ')
 			ct.Options.Default.formatFast(buf)
 		} else {
diff --git a/go/vt/vtadmin/grpcserver/server.go b/go/vt/vtadmin/grpcserver/server.go
index d2667cd745..ea3e469e15 100644
--- a/go/vt/vtadmin/grpcserver/server.go
+++ b/go/vt/vtadmin/grpcserver/server.go
@@ -76,9 +76,6 @@ type Options struct {
 	// See https://github.com/grpc/grpc/blob/7324556353e831c57d30973db33df489c3ed3576/doc/health-checking.md
 	// for more details on healthchecking.
 	Services []string
-
-	StreamInterceptors []grpc.StreamServerInterceptor
-	UnaryInterceptors  []grpc.UnaryServerInterceptor
 }
 
 const healthServiceName = "grpc.health.v1.Health" // reserved health service name
@@ -102,15 +99,9 @@ type Server struct {
 // The underlying gRPC server always has the following interceptors:
 //	- prometheus
 //	- recovery: this handles recovering from panics.
-//
-// The full list of interceptors is as follows:
-// - (optional) interceptors defined on the Options struct
-// - prometheus
-// - (optional) opentracing, if opts.EnableTracing is set
-// - recovery
 func New(name string, opts Options) *Server {
-	streamInterceptors := append(opts.StreamInterceptors, grpc_prometheus.StreamServerInterceptor)
-	unaryInterceptors := append(opts.UnaryInterceptors, grpc_prometheus.UnaryServerInterceptor)
+	streamInterceptors := []grpc.StreamServerInterceptor{grpc_prometheus.StreamServerInterceptor}
+	unaryInterceptors := []grpc.UnaryServerInterceptor{grpc_prometheus.UnaryServerInterceptor}
 
 	if opts.EnableTracing {
 		tracer := opentracing.GlobalTracer()
diff --git a/go/vt/vtgate/engine/cached_size.go b/go/vt/vtgate/engine/cached_size.go
index acfc640f6e..0c1ce058f6 100644
--- a/go/vt/vtgate/engine/cached_size.go
+++ b/go/vt/vtgate/engine/cached_size.go
@@ -562,15 +562,13 @@ func (cached *RevertMigration) CachedSize(alloc bool) int64 {
 	}
 	return size
 }
-
-//go:nocheckptr
 func (cached *Route) CachedSize(alloc bool) int64 {
 	if cached == nil {
 		return int64(0)
 	}
 	size := int64(0)
 	if alloc {
-		size += int64(208)
+		size += int64(224)
 	}
 	// field Keyspace *vitess.io/vitess/go/vt/vtgate/vindexes.Keyspace
 	size += cached.Keyspace.CachedSize(true)
@@ -608,19 +606,11 @@ func (cached *Route) CachedSize(alloc bool) int64 {
 			}
 		}
 	}
-	// field SysTableTableName map[string]vitess.io/vitess/go/vt/vtgate/evalengine.Expr
-	if cached.SysTableTableName != nil {
-		size += int64(48)
-		hmap := reflect.ValueOf(cached.SysTableTableName)
-		numBuckets := int(math.Pow(2, float64((*(*uint8)(unsafe.Pointer(hmap.Pointer() + uintptr(9)))))))
-		numOldBuckets := (*(*uint16)(unsafe.Pointer(hmap.Pointer() + uintptr(10))))
-		size += int64(numOldBuckets * 272)
-		if len(cached.SysTableTableName) > 0 || numBuckets > 1 {
-			size += int64(numBuckets * 272)
-		}
-		for k, v := range cached.SysTableTableName {
-			size += int64(len(k))
-			if cc, ok := v.(cachedObject); ok {
+	// field SysTableTableName []vitess.io/vitess/go/vt/vtgate/evalengine.Expr
+	{
+		size += int64(cap(cached.SysTableTableName)) * int64(16)
+		for _, elem := range cached.SysTableTableName {
+			if cc, ok := elem.(cachedObject); ok {
 				size += cc.CachedSize(true)
 			}
 		}
@@ -680,7 +670,7 @@ func (cached *Send) CachedSize(alloc bool) int64 {
 	}
 	size := int64(0)
 	if alloc {
-		size += int64(44)
+		size += int64(43)
 	}
 	// field Keyspace *vitess.io/vitess/go/vt/vtgate/vindexes.Keyspace
 	size += cached.Keyspace.CachedSize(true)
diff --git a/go/vt/vtgate/engine/route.go b/go/vt/vtgate/engine/route.go
index 42e7d82bf9..cc5df58aea 100644
--- a/go/vt/vtgate/engine/route.go
+++ b/go/vt/vtgate/engine/route.go
@@ -21,7 +21,6 @@ import (
 	"fmt"
 	"sort"
 	"strconv"
-	"strings"
 	"time"
 
 	"vitess.io/vitess/go/vt/log"
@@ -96,7 +95,7 @@ type Route struct {
 
 	// The following two fields are used when routing information_schema queries
 	SysTableTableSchema []evalengine.Expr
-	SysTableTableName   map[string]evalengine.Expr
+	SysTableTableName   []evalengine.Expr
 
 	// Route does not take inputs
 	noInputs
@@ -477,25 +476,32 @@ func (route *Route) routeInfoSchemaQuery(vcursor VCursor, bindVars map[string]*q
 		bindVars[sqltypes.BvSchemaName] = sqltypes.StringBindVariable(specifiedKS)
 	}
 
-	tableNames := map[string]string{}
-	for tblBvName, sysTableName := range route.SysTableTableName {
+	var tableName string
+	for _, sysTableName := range route.SysTableTableName {
 		val, err := sysTableName.Evaluate(env)
 		if err != nil {
 			return nil, err
 		}
 		tabName := val.Value().ToString()
-		tableNames[tblBvName] = tabName
-		bindVars[tblBvName] = sqltypes.StringBindVariable(tabName)
+		if tableName == "" {
+			tableName = tabName
+		}
+		if tableName != tabName {
+			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "two predicates for table_name not supported")
+		}
+	}
+	if tableName != "" {
+		bindVars[BvTableName] = sqltypes.StringBindVariable(tableName)
 	}
 
-	// if the table_schema is system schema, route to default keyspace.
+	// if the table_schema is system system, route to default keyspace.
 	if sqlparser.SystemSchema(specifiedKS) {
 		return defaultRoute()
 	}
 
 	// the use has specified a table_name - let's check if it's a routed table
-	if len(tableNames) > 0 {
-		rss, err := route.paramsRoutedTable(vcursor, bindVars, specifiedKS, tableNames)
+	if tableName != "" {
+		rss, err := route.paramsRoutedTable(vcursor, bindVars, specifiedKS, tableName)
 		if err != nil {
 			// Only if keyspace is not found in vschema, we try with default keyspace.
 			// As the in the table_schema predicates for a keyspace 'ks' it can contain 'vt_ks'.
@@ -525,39 +531,28 @@ func (route *Route) routeInfoSchemaQuery(vcursor VCursor, bindVars map[string]*q
 	return destinations, nil
 }
 
-func (route *Route) paramsRoutedTable(vcursor VCursor, bindVars map[string]*querypb.BindVariable, tableSchema string, tableNames map[string]string) ([]*srvtopo.ResolvedShard, error) {
-	var routedKs *vindexes.Keyspace
-	for tblBvName, tableName := range tableNames {
-		tbl := sqlparser.TableName{
-			Name:      sqlparser.NewTableIdent(tableName),
-			Qualifier: sqlparser.NewTableIdent(tableSchema),
-		}
-		routedTable, err := vcursor.FindRoutedTable(tbl)
-		if err != nil {
-			return nil, err
-		}
-
-		if routedTable != nil {
-			// if we were able to find information about this table, let's use it
-
-			// check if the query is send to single keyspace.
-			if routedKs == nil {
-				routedKs = routedTable.Keyspace
-			}
-			if routedKs.Name != routedTable.Keyspace.Name {
-				return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "cannot send the query to multiple keyspace due to different table_name: %s, %s", routedKs.Name, routedTable.Keyspace.Name)
-			}
+func (route *Route) paramsRoutedTable(vcursor VCursor, bindVars map[string]*querypb.BindVariable, tableSchema string, tableName string) ([]*srvtopo.ResolvedShard, error) {
+	tbl := sqlparser.TableName{
+		Name:      sqlparser.NewTableIdent(tableName),
+		Qualifier: sqlparser.NewTableIdent(tableSchema),
+	}
+	destination, err := vcursor.FindRoutedTable(tbl)
+	if err != nil {
+		return nil, err
+	}
 
-			shards, _, err := vcursor.ResolveDestinations(routedTable.Keyspace.Name, nil, []key.Destination{key.DestinationAnyShard{}})
-			bindVars[tblBvName] = sqltypes.StringBindVariable(routedTable.Name.String())
-			if tableSchema != "" {
-				setReplaceSchemaName(bindVars)
-			}
-			return shards, err
+	if destination != nil {
+		// if we were able to find information about this table, let's use it
+		shards, _, err := vcursor.ResolveDestinations(destination.Keyspace.Name, nil, []key.Destination{key.DestinationAnyShard{}})
+		bindVars[BvTableName] = sqltypes.StringBindVariable(destination.Name.String())
+		if tableSchema != "" {
+			setReplaceSchemaName(bindVars)
 		}
-		// no routed table info found. we'll return nil and check on the outside if we can find the table_schema
-		bindVars[tblBvName] = sqltypes.StringBindVariable(tableName)
+		return shards, err
 	}
+
+	// no routed table info found. we'll return nil and check on the outside if we can find the table_schema
+	bindVars[BvTableName] = sqltypes.StringBindVariable(tableName)
 	return nil, nil
 }
 
@@ -801,12 +796,15 @@ func (route *Route) description() PrimitiveDescription {
 		other["SysTableTableSchema"] = sysTabSchema
 	}
 	if len(route.SysTableTableName) != 0 {
-		var sysTableName []string
-		for k, v := range route.SysTableTableName {
-			sysTableName = append(sysTableName, k+":"+v.String())
+		sysTableName := "["
+		for idx, tableName := range route.SysTableTableName {
+			if idx != 0 {
+				sysTableName += ", "
+			}
+			sysTableName += tableName.String()
 		}
-		sort.Strings(sysTableName)
-		other["SysTableTableName"] = "[" + strings.Join(sysTableName, ", ") + "]"
+		sysTableName += "]"
+		other["SysTableTableName"] = sysTableName
 	}
 	orderBy := GenericJoin(route.OrderBy, orderByToString)
 	if orderBy != "" {
@@ -830,3 +828,6 @@ func (route *Route) description() PrimitiveDescription {
 func orderByToString(in interface{}) string {
 	return in.(OrderByParams).String()
 }
+
+// BvTableName is used to fill in the table name for information_schema queries with routed tables
+const BvTableName = "__vttablename"
diff --git a/go/vt/vtgate/planbuilder/builder.go b/go/vt/vtgate/planbuilder/builder.go
index 2f61da958e..5260ff184c 100644
--- a/go/vt/vtgate/planbuilder/builder.go
+++ b/go/vt/vtgate/planbuilder/builder.go
@@ -88,12 +88,12 @@ type truncater interface {
 
 // TestBuilder builds a plan for a query based on the specified vschema.
 // This method is only used from tests
-func TestBuilder(query string, vschema ContextVSchema, keyspace string) (*engine.Plan, error) {
+func TestBuilder(query string, vschema ContextVSchema) (*engine.Plan, error) {
 	stmt, reserved, err := sqlparser.Parse2(query)
 	if err != nil {
 		return nil, err
 	}
-	result, err := sqlparser.RewriteAST(stmt, keyspace)
+	result, err := sqlparser.RewriteAST(stmt, "")
 	if err != nil {
 		return nil, err
 	}
diff --git a/go/vt/vtgate/planbuilder/jointree.go b/go/vt/vtgate/planbuilder/jointree.go
index 9ea7c5c07d..8aedd07e00 100644
--- a/go/vt/vtgate/planbuilder/jointree.go
+++ b/go/vt/vtgate/planbuilder/jointree.go
@@ -19,8 +19,6 @@ package planbuilder
 import (
 	"strings"
 
-	"vitess.io/vitess/go/vt/vtgate/evalengine"
-
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vtgate/engine"
@@ -97,10 +95,6 @@ type (
 
 		// columns needed to feed other plans`
 		columns []*sqlparser.ColName
-
-		// The following two fields are used when routing information_schema queries
-		SysTableTableSchema []evalengine.Expr
-		SysTableTableName   map[string]evalengine.Expr
 	}
 
 	joinPlan struct {
diff --git a/go/vt/vtgate/planbuilder/jointree_transformers.go b/go/vt/vtgate/planbuilder/jointree_transformers.go
index 44aae7f5f4..dfa98d649b 100644
--- a/go/vt/vtgate/planbuilder/jointree_transformers.go
+++ b/go/vt/vtgate/planbuilder/jointree_transformers.go
@@ -144,13 +144,11 @@ func transformRoutePlan(n *routePlan) (*route, error) {
 
 	return &route{
 		eroute: &engine.Route{
-			Opcode:              n.routeOpCode,
-			TableName:           strings.Join(tableNames, ", "),
-			Keyspace:            n.keyspace,
-			Vindex:              singleColumn,
-			Values:              n.vindexValues,
-			SysTableTableName:   n.SysTableTableName,
-			SysTableTableSchema: n.SysTableTableSchema,
+			Opcode:    n.routeOpCode,
+			TableName: strings.Join(tableNames, ", "),
+			Keyspace:  n.keyspace,
+			Vindex:    singleColumn,
+			Values:    n.vindexValues,
 		},
 		Select: &sqlparser.Select{
 			SelectExprs: expressions,
diff --git a/go/vt/vtgate/planbuilder/route_planning.go b/go/vt/vtgate/planbuilder/route_planning.go
index 92033cd6a3..91903c2af3 100644
--- a/go/vt/vtgate/planbuilder/route_planning.go
+++ b/go/vt/vtgate/planbuilder/route_planning.go
@@ -41,38 +41,21 @@ func gen4Planner(_ string) func(sqlparser.Statement, *sqlparser.ReservedVars, Co
 		if !ok {
 			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "%T not yet supported", stmt)
 		}
-
-		getPlan := func(sel *sqlparser.Select) (logicalPlan, error) {
-			return newBuildSelectPlan(sel, reservedVars, vschema)
-		}
-
-		plan, err := getPlan(sel)
-		if err != nil {
-			return nil, err
-		}
-
-		if shouldRetryWithCNFRewriting(plan) {
-			// by transforming the predicates to CNF, the planner will sometimes find better plans
-			primitive := rewriteToCNFAndReplan(stmt, getPlan)
-			if primitive != nil {
-				return primitive, nil
-			}
-		}
-		return plan.Primitive(), nil
+		return newBuildSelectPlan(sel, vschema)
 	}
 }
 
-func newBuildSelectPlan(sel *sqlparser.Select, reservedVars *sqlparser.ReservedVars, vschema ContextVSchema) (logicalPlan, error) {
+func newBuildSelectPlan(sel *sqlparser.Select, vschema ContextVSchema) (engine.Primitive, error) {
 
 	directives := sqlparser.ExtractCommentDirectives(sel.Comments)
 	if len(directives) > 0 {
 		return nil, semantics.Gen4NotSupportedF("comment directives")
 	}
-	ksName := ""
-	if ks, _ := vschema.DefaultKeyspace(); ks != nil {
-		ksName = ks.Name
+	keyspace, err := vschema.DefaultKeyspace()
+	if err != nil {
+		return nil, err
 	}
-	semTable, err := semantics.Analyze(sel, ksName, vschema)
+	semTable, err := semantics.Analyze(sel, keyspace.Name, vschema)
 	if err != nil {
 		return nil, err
 	}
@@ -87,7 +70,7 @@ func newBuildSelectPlan(sel *sqlparser.Select, reservedVars *sqlparser.ReservedV
 		return nil, err
 	}
 
-	tree, err := optimizeQuery(opTree, reservedVars, semTable, vschema)
+	tree, err := optimizeQuery(opTree, semTable, vschema)
 	if err != nil {
 		return nil, err
 	}
@@ -117,35 +100,34 @@ func newBuildSelectPlan(sel *sqlparser.Select, reservedVars *sqlparser.ReservedV
 	if err := plan.WireupGen4(semTable); err != nil {
 		return nil, err
 	}
-
-	return plan, nil
+	return plan.Primitive(), nil
 }
 
-func optimizeQuery(opTree abstract.Operator, reservedVars *sqlparser.ReservedVars, semTable *semantics.SemTable, vschema ContextVSchema) (joinTree, error) {
+func optimizeQuery(opTree abstract.Operator, semTable *semantics.SemTable, vschema ContextVSchema) (joinTree, error) {
 	switch op := opTree.(type) {
 	case *abstract.QueryGraph:
 		switch {
 		case vschema.Planner() == Gen4Left2Right:
-			return leftToRightSolve(op, reservedVars, semTable, vschema)
+			return leftToRightSolve(op, semTable, vschema)
 		default:
-			return greedySolve(op, reservedVars, semTable, vschema)
+			return greedySolve(op, semTable, vschema)
 		}
 	case *abstract.LeftJoin:
-		treeInner, err := optimizeQuery(op.Left, reservedVars, semTable, vschema)
+		treeInner, err := optimizeQuery(op.Left, semTable, vschema)
 		if err != nil {
 			return nil, err
 		}
-		treeOuter, err := optimizeQuery(op.Right, reservedVars, semTable, vschema)
+		treeOuter, err := optimizeQuery(op.Right, semTable, vschema)
 		if err != nil {
 			return nil, err
 		}
 		return mergeOrJoin(treeInner, treeOuter, []sqlparser.Expr{op.Predicate}, semTable, false)
 	case *abstract.Join:
-		treeInner, err := optimizeQuery(op.LHS, reservedVars, semTable, vschema)
+		treeInner, err := optimizeQuery(op.LHS, semTable, vschema)
 		if err != nil {
 			return nil, err
 		}
-		treeOuter, err := optimizeQuery(op.RHS, reservedVars, semTable, vschema)
+		treeOuter, err := optimizeQuery(op.RHS, semTable, vschema)
 		if err != nil {
 			return nil, err
 		}
@@ -442,8 +424,8 @@ type (
 	and removes the two inputs to this cheapest plan and instead adds the join.
 	As an optimization, it first only considers joining tables that have predicates defined between them
 */
-func greedySolve(qg *abstract.QueryGraph, reservedVars *sqlparser.ReservedVars, semTable *semantics.SemTable, vschema ContextVSchema) (joinTree, error) {
-	joinTrees, err := seedPlanList(qg, reservedVars, semTable, vschema)
+func greedySolve(qg *abstract.QueryGraph, semTable *semantics.SemTable, vschema ContextVSchema) (joinTree, error) {
+	joinTrees, err := seedPlanList(qg, semTable, vschema)
 	planCache := cacheMap{}
 	if err != nil {
 		return nil, err
@@ -535,8 +517,8 @@ func findBestJoinTree(
 	return bestPlan, lIdx, rIdx, nil
 }
 
-func leftToRightSolve(qg *abstract.QueryGraph, reservedVars *sqlparser.ReservedVars, semTable *semantics.SemTable, vschema ContextVSchema) (joinTree, error) {
-	plans, err := seedPlanList(qg, reservedVars, semTable, vschema)
+func leftToRightSolve(qg *abstract.QueryGraph, semTable *semantics.SemTable, vschema ContextVSchema) (joinTree, error) {
+	plans, err := seedPlanList(qg, semTable, vschema)
 	if err != nil {
 		return nil, err
 	}
@@ -558,13 +540,13 @@ func leftToRightSolve(qg *abstract.QueryGraph, reservedVars *sqlparser.ReservedV
 }
 
 // seedPlanList returns a routePlan for each table in the qg
-func seedPlanList(qg *abstract.QueryGraph, reservedVars *sqlparser.ReservedVars, semTable *semantics.SemTable, vschema ContextVSchema) ([]joinTree, error) {
+func seedPlanList(qg *abstract.QueryGraph, semTable *semantics.SemTable, vschema ContextVSchema) ([]joinTree, error) {
 	plans := make([]joinTree, len(qg.Tables))
 
 	// we start by seeding the table with the single routes
 	for i, table := range qg.Tables {
 		solves := semTable.TableSetFor(table.Alias)
-		plan, err := createRoutePlan(table, solves, reservedVars, vschema)
+		plan, err := createRoutePlan(table, solves, vschema)
 		if err != nil {
 			return nil, err
 		}
@@ -580,31 +562,26 @@ func removeAt(plans []joinTree, idx int) []joinTree {
 	return append(plans[:idx], plans[idx+1:]...)
 }
 
-func createRoutePlan(table *abstract.QueryTable, solves semantics.TableSet, reservedVars *sqlparser.ReservedVars, vschema ContextVSchema) (*routePlan, error) {
+func createRoutePlan(table *abstract.QueryTable, solves semantics.TableSet, vschema ContextVSchema) (*routePlan, error) {
 	if table.IsInfSchema {
-		ks, err := vschema.AnyKeyspace()
+		defaultKeyspace, err := vschema.DefaultKeyspace()
 		if err != nil {
 			return nil, err
 		}
-		rp := &routePlan{
+		return &routePlan{
 			routeOpCode: engine.SelectDBA,
 			solved:      solves,
-			keyspace:    ks,
+			// TODO: find keyspace to route using the predicates as in v3
+			keyspace: defaultKeyspace,
 			tables: []relation{&routeTable{
 				qtable: table,
 				vtable: &vindexes.Table{
 					Name:     table.Table.Name,
-					Keyspace: ks,
+					Keyspace: defaultKeyspace,
 				},
 			}},
 			predicates: table.Predicates,
-		}
-		err = rp.findSysInfoRoutingPredicatesGen4(reservedVars)
-		if err != nil {
-			return nil, err
-		}
-
-		return rp, nil
+		}, nil
 	}
 	vschemaTable, _, _, _, _, err := vschema.FindTableOrVindex(table.Table)
 	if err != nil {
@@ -729,11 +706,6 @@ func tryMerge(a, b joinTree, joinPredicates []sqlparser.Expr, semTable *semantic
 	if aRoute == nil || bRoute == nil {
 		return nil
 	}
-	// If both the routes are system schema queries then we do not validate the keyspaces at plan time. As they are not always the ones where the query will be sent to.
-	if (aRoute.routeOpCode != engine.SelectDBA ||
-		bRoute.routeOpCode != engine.SelectDBA) && aRoute.keyspace != bRoute.keyspace {
-		return nil
-	}
 
 	newTabletSet := aRoute.solved | bRoute.solved
 
@@ -782,6 +754,9 @@ func joinTreesToRoutes(a, b joinTree) (*routePlan, *routePlan) {
 	if !ok {
 		return nil, nil
 	}
+	if aRoute.keyspace != bRoute.keyspace {
+		return nil, nil
+	}
 	return aRoute, bRoute
 }
 
@@ -792,17 +767,6 @@ func createRoutePlanForInner(aRoute *routePlan, bRoute *routePlan, newTabletSet
 	} else {
 		tables = append(parenTables{aRoute.tables}, bRoute.tables...)
 	}
-
-	// append system table names from both the routes.
-	sysTableName := aRoute.SysTableTableName
-	if sysTableName == nil {
-		sysTableName = bRoute.SysTableTableName
-	} else {
-		for k, v := range bRoute.SysTableTableName {
-			sysTableName[k] = v
-		}
-	}
-
 	return &routePlan{
 		routeOpCode: aRoute.routeOpCode,
 		solved:      newTabletSet,
@@ -810,11 +774,9 @@ func createRoutePlanForInner(aRoute *routePlan, bRoute *routePlan, newTabletSet
 		predicates: append(
 			append(aRoute.predicates, bRoute.predicates...),
 			joinPredicates...),
-		keyspace:            aRoute.keyspace,
-		vindexPreds:         append(aRoute.vindexPreds, bRoute.vindexPreds...),
-		leftJoins:           append(aRoute.leftJoins, bRoute.leftJoins...),
-		SysTableTableSchema: append(aRoute.SysTableTableSchema, bRoute.SysTableTableSchema...),
-		SysTableTableName:   sysTableName,
+		keyspace:    aRoute.keyspace,
+		vindexPreds: append(aRoute.vindexPreds, bRoute.vindexPreds...),
+		leftJoins:   append(aRoute.leftJoins, bRoute.leftJoins...),
 	}
 }
 
diff --git a/go/vt/vtgate/planbuilder/select.go b/go/vt/vtgate/planbuilder/select.go
index 44bb541a9b..3a7cba1cd5 100644
--- a/go/vt/vtgate/planbuilder/select.go
+++ b/go/vt/vtgate/planbuilder/select.go
@@ -371,7 +371,7 @@ func (pb *primitiveBuilder) pushFilter(in sqlparser.Expr, whereType string, rese
 		}
 		rut, isRoute := origin.(*route)
 		if isRoute && rut.eroute.Opcode == engine.SelectDBA {
-			err := pb.findSysInfoRoutingPredicates(expr, rut, reservedVars)
+			err := pb.findSysInfoRoutingPredicates(expr, rut)
 			if err != nil {
 				return err
 			}
diff --git a/go/vt/vtgate/planbuilder/system_tables.go b/go/vt/vtgate/planbuilder/system_tables.go
index 136918ff1f..111d86bb41 100644
--- a/go/vt/vtgate/planbuilder/system_tables.go
+++ b/go/vt/vtgate/planbuilder/system_tables.go
@@ -19,11 +19,12 @@ package planbuilder
 import (
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/vtgate/engine"
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
 )
 
-func (pb *primitiveBuilder) findSysInfoRoutingPredicates(expr sqlparser.Expr, rut *route, reservedVars *sqlparser.ReservedVars) error {
-	isTableSchema, bvName, out, err := extractInfoSchemaRoutingPredicate(expr, reservedVars)
+func (pb *primitiveBuilder) findSysInfoRoutingPredicates(expr sqlparser.Expr, rut *route) error {
+	isTableSchema, out, err := extractInfoSchemaRoutingPredicate(expr)
 	if err != nil {
 		return err
 	}
@@ -35,38 +36,12 @@ func (pb *primitiveBuilder) findSysInfoRoutingPredicates(expr sqlparser.Expr, ru
 	if isTableSchema {
 		rut.eroute.SysTableTableSchema = append(rut.eroute.SysTableTableSchema, out)
 	} else {
-		if rut.eroute.SysTableTableName == nil {
-			rut.eroute.SysTableTableName = map[string]evalengine.Expr{}
-		}
-		rut.eroute.SysTableTableName[bvName] = out
+		rut.eroute.SysTableTableName = append(rut.eroute.SysTableTableName, out)
 	}
 
 	return nil
 }
 
-func (rp *routePlan) findSysInfoRoutingPredicatesGen4(reservedVars *sqlparser.ReservedVars) error {
-	for _, pred := range rp.predicates {
-		isTableSchema, bvName, out, err := extractInfoSchemaRoutingPredicate(pred, reservedVars)
-		if err != nil {
-			return err
-		}
-		if out == nil {
-			// we didn't find a predicate to use for routing, continue to look for next predicate
-			continue
-		}
-
-		if isTableSchema {
-			rp.SysTableTableSchema = append(rp.SysTableTableSchema, out)
-		} else {
-			if rp.SysTableTableName == nil {
-				rp.SysTableTableName = map[string]evalengine.Expr{}
-			}
-			rp.SysTableTableName[bvName] = out
-		}
-	}
-	return nil
-}
-
 func findOtherComparator(cmp *sqlparser.ComparisonExpr) (bool, sqlparser.Expr, sqlparser.Expr, func(arg sqlparser.Argument)) {
 	if schema, table := isTableSchemaOrName(cmp.Left); schema || table {
 		return schema, cmp.Left, cmp.Right, func(arg sqlparser.Argument) {
@@ -98,7 +73,7 @@ func isTableNameCol(col *sqlparser.ColName) bool {
 	return col.Name.EqualString("table_name")
 }
 
-func extractInfoSchemaRoutingPredicate(in sqlparser.Expr, reservedVars *sqlparser.ReservedVars) (bool, string, evalengine.Expr, error) {
+func extractInfoSchemaRoutingPredicate(in sqlparser.Expr) (bool, evalengine.Expr, error) {
 	switch cmp := in.(type) {
 	case *sqlparser.ComparisonExpr:
 		if cmp.Operator == sqlparser.EqualOp {
@@ -109,22 +84,22 @@ func extractInfoSchemaRoutingPredicate(in sqlparser.Expr, reservedVars *sqlparse
 					if err == sqlparser.ErrExprNotSupported {
 						// This just means we can't rewrite this particular expression,
 						// not that we have to exit altogether
-						return false, "", nil, nil
+						return false, nil, nil
 					}
-					return false, "", nil, err
+					return false, nil, err
 				}
 				var name string
 				if isSchemaName {
 					name = sqltypes.BvSchemaName
 				} else {
-					name = reservedVars.ReserveColName(col.(*sqlparser.ColName))
+					name = engine.BvTableName
 				}
 				replaceOther(sqlparser.NewArgument(name))
-				return isSchemaName, name, evalExpr, nil
+				return isSchemaName, evalExpr, nil
 			}
 		}
 	}
-	return false, "", nil, nil
+	return false, nil, nil
 }
 
 func shouldRewrite(e sqlparser.Expr) bool {


Test patch:
diff --git a/go/mysql/auth_server_clientcert_test.go b/go/mysql/auth_server_clientcert_test.go
index 0922858212..468f675e55 100644
--- a/go/mysql/auth_server_clientcert_test.go
+++ b/go/mysql/auth_server_clientcert_test.go
@@ -35,9 +35,7 @@ const clientCertUsername = "Client Cert"
 func TestValidCert(t *testing.T) {
 	th := &testHandler{}
 
-	authServer := &AuthServerClientCert{
-		Method: MysqlClearPassword,
-	}
+	authServer := newAuthServerClientCert()
 
 	// Create the listener, so we can get its host.
 	l, err := NewListener("tcp", ":0", authServer, th, 0, 0, false)
@@ -120,9 +118,7 @@ func TestValidCert(t *testing.T) {
 func TestNoCert(t *testing.T) {
 	th := &testHandler{}
 
-	authServer := &AuthServerClientCert{
-		Method: MysqlClearPassword,
-	}
+	authServer := newAuthServerClientCert()
 
 	// Create the listener, so we can get its host.
 	l, err := NewListener("tcp", ":0", authServer, th, 0, 0, false)
diff --git a/go/mysql/auth_server_static_test.go b/go/mysql/auth_server_static_test.go
index d5e43f312f..6a0170d492 100644
--- a/go/mysql/auth_server_static_test.go
+++ b/go/mysql/auth_server_static_test.go
@@ -87,13 +87,13 @@ func TestValidateHashGetter(t *testing.T) {
 	ip := net.ParseIP("127.0.0.1")
 	addr := &net.IPAddr{IP: ip, Zone: ""}
 
-	salt, err := NewSalt()
+	salt, err := newSalt()
 	if err != nil {
 		t.Fatalf("error generating salt: %v", err)
 	}
 
 	scrambled := ScrambleMysqlNativePassword(salt, []byte("password"))
-	getter, err := auth.ValidateHash(salt, "mysql_user", scrambled, addr)
+	getter, err := auth.UserEntryWithHash(nil, salt, "mysql_user", scrambled, addr)
 	if err != nil {
 		t.Fatalf("error validating password: %v", err)
 	}
@@ -150,6 +150,8 @@ func TestStaticConfigHUP(t *testing.T) {
 	hupTest(t, aStatic, tmpFile, oldStr, "str2")
 	hupTest(t, aStatic, tmpFile, "str2", "str3") // still handling the signal
 
+	mu.Lock()
+	defer mu.Unlock()
 	// delete registered Auth server
 	for auth := range authServers {
 		delete(authServers, auth)
@@ -265,13 +267,13 @@ func TestStaticPasswords(t *testing.T) {
 
 	for _, c := range tests {
 		t.Run(fmt.Sprintf("%s-%s", c.user, c.password), func(t *testing.T) {
-			salt, err := NewSalt()
+			salt, err := newSalt()
 			if err != nil {
 				t.Fatalf("error generating salt: %v", err)
 			}
 
 			scrambled := ScrambleMysqlNativePassword(salt, []byte(c.password))
-			_, err = auth.ValidateHash(salt, c.user, scrambled, addr)
+			_, err = auth.UserEntryWithHash(nil, salt, c.user, scrambled, addr)
 
 			if c.success {
 				if err != nil {
diff --git a/go/mysql/auth_server_test.go b/go/mysql/auth_server_test.go
new file mode 100644
index 0000000000..14457da8ba
--- /dev/null
+++ b/go/mysql/auth_server_test.go
@@ -0,0 +1,55 @@
+/*
+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 mysql
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestVerifyHashedMysqlNativePassword(t *testing.T) {
+	salt := []byte{10, 47, 74, 111, 75, 73, 34, 48, 88, 76, 114, 74, 37, 13, 3, 80, 82, 2, 23, 21}
+	password := "secret"
+
+	// Double SHA1 of "secret"
+	passwordHash := []byte{0x38, 0x81, 0x21, 0x9d, 0x08, 0x7d, 0xd9, 0xc6, 0x34, 0x37, 0x3f, 0xd3,
+		0x3d, 0xfa, 0x33, 0xa2, 0xcb, 0x6b, 0xfc, 0x6c, 0x52, 0x0b, 0x64, 0xb8,
+		0xbb, 0x60, 0xef, 0x2c, 0xeb, 0x53, 0x4a, 0xe7}
+
+	reply := ScrambleCachingSha2Password(salt, []byte(password))
+
+	assert.True(t, VerifyHashedCachingSha2Password(reply, salt, passwordHash), "password hash mismatch")
+
+	passwordHash[0] = 0x00
+	assert.False(t, VerifyHashedCachingSha2Password(reply, salt, passwordHash), "password hash match")
+}
+
+func TestVerifyHashedCachingSha2Password(t *testing.T) {
+	salt := []byte{10, 47, 74, 111, 75, 73, 34, 48, 88, 76, 114, 74, 37, 13, 3, 80, 82, 2, 23, 21}
+	password := "secret"
+
+	// Double SHA1 of "secret"
+	passwordHash := []byte{0x14, 0xe6, 0x55, 0x67, 0xab, 0xdb, 0x51, 0x35, 0xd0, 0xcf, 0xd9, 0xa7,
+		0x0b, 0x30, 0x32, 0xc1, 0x79, 0xa4, 0x9e, 0xe7}
+
+	reply := ScrambleMysqlNativePassword(salt, []byte(password))
+	assert.True(t, VerifyHashedMysqlNativePassword(reply, salt, passwordHash), "password hash mismatch")
+
+	passwordHash[0] = 0x00
+	assert.False(t, VerifyHashedMysqlNativePassword(reply, salt, passwordHash), "password hash match")
+}
diff --git a/go/mysql/handshake_test.go b/go/mysql/handshake_test.go
index 35f3224d63..8acc543a6f 100644
--- a/go/mysql/handshake_test.go
+++ b/go/mysql/handshake_test.go
@@ -38,8 +38,7 @@ import (
 func TestClearTextClientAuth(t *testing.T) {
 	th := &testHandler{}
 
-	authServer := NewAuthServerStatic("", "", 0)
-	authServer.method = MysqlClearPassword
+	authServer := NewAuthServerStaticWithAuthMethodDescription("", "", 0, MysqlClearPassword)
 	authServer.entries["user1"] = []*AuthServerStaticEntry{
 		{Password: "password1"},
 	}
@@ -96,7 +95,7 @@ func TestClearTextClientAuth(t *testing.T) {
 func TestSSLConnection(t *testing.T) {
 	th := &testHandler{}
 
-	authServer := NewAuthServerStatic("", "", 0)
+	authServer := NewAuthServerStaticWithAuthMethodDescription("", "", 0, MysqlClearPassword)
 	authServer.entries["user1"] = []*AuthServerStaticEntry{
 		{Password: "password1"},
 	}
@@ -156,7 +155,6 @@ func TestSSLConnection(t *testing.T) {
 
 	// Make sure clear text auth works over SSL.
 	t.Run("ClearText", func(t *testing.T) {
-		authServer.method = MysqlClearPassword
 		testSSLConnectionClearText(t, params)
 	})
 }
diff --git a/go/mysql/query_benchmark_test.go b/go/mysql/query_benchmark_test.go
index 155d809675..7b9a8c2d8f 100644
--- a/go/mysql/query_benchmark_test.go
+++ b/go/mysql/query_benchmark_test.go
@@ -37,7 +37,7 @@ const benchmarkQueryPrefix = "benchmark "
 func benchmarkQuery(b *testing.B, threads int, query string) {
 	th := &testHandler{}
 
-	authServer := &AuthServerNone{}
+	authServer := NewAuthServerNone()
 
 	lCfg := ListenerConfig{
 		Protocol:           "tcp",
diff --git a/go/mysql/server_test.go b/go/mysql/server_test.go
index a322314a5a..fa1fe9a198 100644
--- a/go/mysql/server_test.go
+++ b/go/mysql/server_test.go
@@ -34,6 +34,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"vitess.io/vitess/go/sqltypes"
+	"vitess.io/vitess/go/test/utils"
 	vtenv "vitess.io/vitess/go/vt/env"
 	"vitess.io/vitess/go/vt/tlstest"
 	"vitess.io/vitess/go/vt/vterrors"
@@ -694,12 +695,11 @@ func TestServerStats(t *testing.T) {
 func TestClearTextServer(t *testing.T) {
 	th := &testHandler{}
 
-	authServer := NewAuthServerStatic("", "", 0)
+	authServer := NewAuthServerStaticWithAuthMethodDescription("", "", 0, MysqlClearPassword)
 	authServer.entries["user1"] = []*AuthServerStaticEntry{{
 		Password: "password1",
 		UserData: "userData1",
 	}}
-	authServer.method = MysqlClearPassword
 	defer authServer.close()
 	l, err := NewListener("tcp", ":0", authServer, th, 0, 0, false)
 	require.NoError(t, err)
@@ -768,12 +768,11 @@ func TestClearTextServer(t *testing.T) {
 func TestDialogServer(t *testing.T) {
 	th := &testHandler{}
 
-	authServer := NewAuthServerStatic("", "", 0)
+	authServer := NewAuthServerStaticWithAuthMethodDescription("", "", 0, MysqlDialog)
 	authServer.entries["user1"] = []*AuthServerStaticEntry{{
 		Password: "password1",
 		UserData: "userData1",
 	}}
-	authServer.method = MysqlDialog
 	defer authServer.close()
 	l, err := NewListener("tcp", ":0", authServer, th, 0, 0, false)
 	require.NoError(t, err)
@@ -987,6 +986,120 @@ func TestTLSRequired(t *testing.T) {
 	}
 }
 
+func TestCachingSha2PasswordAuthWithTLS(t *testing.T) {
+	th := &testHandler{}
+
+	authServer := NewAuthServerStaticWithAuthMethodDescription("", "", 0, CachingSha2Password)
+	authServer.entries["user1"] = []*AuthServerStaticEntry{
+		{Password: "password1"},
+	}
+	defer authServer.close()
+
+	// Create the listener, so we can get its host.
+	l, err := NewListener("tcp", ":0", authServer, th, 0, 0, false)
+	if err != nil {
+		t.Fatalf("NewListener failed: %v", err)
+	}
+	defer l.Close()
+	host := l.Addr().(*net.TCPAddr).IP.String()
+	port := l.Addr().(*net.TCPAddr).Port
+
+	// Create the certs.
+	root, err := ioutil.TempDir("", "TestSSLConnection")
+	if err != nil {
+		t.Fatalf("TempDir failed: %v", err)
+	}
+	defer os.RemoveAll(root)
+	tlstest.CreateCA(root)
+	tlstest.CreateSignedCert(root, tlstest.CA, "01", "server", "server.example.com")
+	tlstest.CreateSignedCert(root, tlstest.CA, "02", "client", "Client Cert")
+
+	// Create the server with TLS config.
+	serverConfig, err := vttls.ServerConfig(
+		path.Join(root, "server-cert.pem"),
+		path.Join(root, "server-key.pem"),
+		path.Join(root, "ca-cert.pem"),
+		"",
+		tls.VersionTLS12)
+	if err != nil {
+		t.Fatalf("TLSServerConfig failed: %v", err)
+	}
+	l.TLSConfig.Store(serverConfig)
+	go func() {
+		l.Accept()
+	}()
+
+	// Setup the right parameters.
+	params := &ConnParams{
+		Host:  host,
+		Port:  port,
+		Uname: "user1",
+		Pass:  "password1",
+		// SSL flags.
+		Flags:      CapabilityClientSSL,
+		SslCa:      path.Join(root, "ca-cert.pem"),
+		SslCert:    path.Join(root, "client-cert.pem"),
+		SslKey:     path.Join(root, "client-key.pem"),
+		ServerName: "server.example.com",
+	}
+
+	// Connection should fail, as server requires SSL for caching_sha2_password.
+	ctx := context.Background()
+
+	conn, err := Connect(ctx, params)
+	if err != nil {
+		t.Fatalf("unexpected connection error: %v", err)
+	}
+	defer conn.Close()
+
+	// Run a 'select rows' command with results.
+	result, err := conn.ExecuteFetch("select rows", 10000, true)
+	if err != nil {
+		t.Fatalf("ExecuteFetch failed: %v", err)
+	}
+	utils.MustMatch(t, result, selectRowsResult)
+
+	// Send a ComQuit to avoid the error message on the server side.
+	conn.writeComQuit()
+}
+
+func TestCachingSha2PasswordAuthWithoutTLS(t *testing.T) {
+	th := &testHandler{}
+
+	authServer := NewAuthServerStaticWithAuthMethodDescription("", "", 0, CachingSha2Password)
+	authServer.entries["user1"] = []*AuthServerStaticEntry{
+		{Password: "password1"},
+	}
+	defer authServer.close()
+
+	// Create the listener.
+	l, err := NewListener("tcp", ":0", authServer, th, 0, 0, false)
+	if err != nil {
+		t.Fatalf("NewListener failed: %v", err)
+	}
+	defer l.Close()
+	host := l.Addr().(*net.TCPAddr).IP.String()
+	port := l.Addr().(*net.TCPAddr).Port
+	go func() {
+		l.Accept()
+	}()
+
+	// Setup the right parameters.
+	params := &ConnParams{
+		Host:  host,
+		Port:  port,
+		Uname: "user1",
+		Pass:  "password1",
+	}
+
+	// Connection should fail, as server requires SSL for caching_sha2_password.
+	ctx := context.Background()
+	_, err = Connect(ctx, params)
+	if err == nil || !strings.Contains(err.Error(), "No authentication methods available for authentication") {
+		t.Fatalf("unexpected connection error: %v", err)
+	}
+}
+
 func checkCountForTLSVer(t *testing.T, version string, expected int64) {
 	connCounts := connCountByTLSVer.Counts()
 	count, ok := connCounts[version]
@@ -1254,7 +1367,7 @@ func TestServerFlush(t *testing.T) {
 
 	th := &testHandler{}
 
-	l, err := NewListener("tcp", ":0", &AuthServerNone{}, th, 0, 0, false)
+	l, err := NewListener("tcp", ":0", NewAuthServerNone(), th, 0, 0, false)
 	require.NoError(t, err)
 	defer l.Close()
 	go l.Accept()
diff --git a/go/test/endtoend/vtgate/gen4/main_test.go b/go/test/endtoend/vtgate/gen4/main_test.go
index 4969d5d8e5..d892042ae1 100644
--- a/go/test/endtoend/vtgate/gen4/main_test.go
+++ b/go/test/endtoend/vtgate/gen4/main_test.go
@@ -77,15 +77,6 @@ create table t2(
     }
   }
 }`
-
-	routingRules = `
-{"rules": [
-  {
-    "from_table": "ks.t1000",
-	"to_tables": ["ks.t1"]
-  }
-]}
-`
 )
 
 func TestMain(m *testing.M) {
@@ -113,17 +104,6 @@ func TestMain(m *testing.M) {
 			return 1
 		}
 
-		// apply routing rules
-		err = clusterInstance.VtctlclientProcess.ApplyRoutingRules(routingRules)
-		if err != nil {
-			return 1
-		}
-
-		err = clusterInstance.VtctlclientProcess.ExecuteCommand("RebuildVSchemaGraph")
-		if err != nil {
-			return 1
-		}
-
 		// Start vtgate
 		clusterInstance.VtGateExtraArgs = []string{"-planner_version", "Gen4"} // enable Gen4 planner.
 		err = clusterInstance.StartVtgate()
diff --git a/go/test/endtoend/vtgate/gen4/system_schema_test.go b/go/test/endtoend/vtgate/gen4/system_schema_test.go
deleted file mode 100644
index cbd3cd2489..0000000000
--- a/go/test/endtoend/vtgate/gen4/system_schema_test.go
+++ /dev/null
@@ -1,202 +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 vtgate
-
-import (
-	"context"
-	"fmt"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-
-	"github.com/stretchr/testify/require"
-
-	"vitess.io/vitess/go/mysql"
-	"vitess.io/vitess/go/test/endtoend/cluster"
-)
-
-func TestDbNameOverride(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.Nil(t, err)
-	defer conn.Close()
-	qr, err := conn.ExecuteFetch("SELECT distinct database() FROM information_schema.tables WHERE table_schema = database()", 1000, true)
-
-	require.Nil(t, err)
-	assert.Equal(t, 1, len(qr.Rows), "did not get enough rows back")
-	assert.Equal(t, "vt_ks", qr.Rows[0][0].ToString())
-}
-
-func TestInformationSchemaQuery(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	assertSingleRowIsReturned(t, conn, "table_schema = 'ks'", "vt_ks")
-	assertSingleRowIsReturned(t, conn, "table_schema = 'vt_ks'", "vt_ks")
-	assertResultIsEmpty(t, conn, "table_schema = 'NONE'")
-	assertSingleRowIsReturned(t, conn, "table_schema = 'performance_schema'", "performance_schema")
-	assertResultIsEmpty(t, conn, "table_schema = 'PERFORMANCE_SCHEMA'")
-	assertSingleRowIsReturned(t, conn, "table_schema = 'performance_schema' and table_name = 'users'", "performance_schema")
-	assertResultIsEmpty(t, conn, "table_schema = 'performance_schema' and table_name = 'foo'")
-	assertSingleRowIsReturned(t, conn, "table_schema = 'vt_ks' and table_name = 't1'", "vt_ks")
-	assertSingleRowIsReturned(t, conn, "table_schema = 'ks' and table_name = 't1'", "vt_ks")
-}
-
-func assertResultIsEmpty(t *testing.T, conn *mysql.Conn, pre string) {
-	t.Run(pre, func(t *testing.T) {
-		qr, err := conn.ExecuteFetch("SELECT distinct table_schema FROM information_schema.tables WHERE "+pre, 1000, true)
-		require.NoError(t, err)
-		assert.Empty(t, qr.Rows)
-	})
-}
-
-func assertSingleRowIsReturned(t *testing.T, conn *mysql.Conn, predicate string, expectedKs string) {
-	t.Run(predicate, func(t *testing.T) {
-		qr, err := conn.ExecuteFetch("SELECT distinct table_schema FROM information_schema.tables WHERE "+predicate, 1000, true)
-		require.NoError(t, err)
-		assert.Equal(t, 1, len(qr.Rows), "did not get enough rows back")
-		assert.Equal(t, expectedKs, qr.Rows[0][0].ToString())
-	})
-}
-
-func TestInformationSchemaWithSubquery(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	result := checkedExec(t, conn, "SELECT column_name FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = (SELECT SCHEMA()) AND TABLE_NAME = 'not_exists'")
-	assert.Empty(t, result.Rows)
-}
-
-func TestInformationSchemaQueryGetsRoutedToTheRightTableAndKeyspace(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	_ = checkedExec(t, conn, "SELECT id FROM t1000") // test that the routed table is available to us
-	result := checkedExec(t, conn, "SELECT * FROM information_schema.tables WHERE table_schema = database() and table_name='t1000'")
-	assert.NotEmpty(t, result.Rows)
-}
-
-func TestFKConstraintUsingInformationSchema(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	checkedExec(t, conn, "create table t7_xxhash(uid varchar(50),phone bigint,msg varchar(100),primary key(uid)) Engine=InnoDB")
-	checkedExec(t, conn, "create table t7_fk(id bigint,t7_uid varchar(50),primary key(id),CONSTRAINT t7_fk_ibfk_1 foreign key (t7_uid) references t7_xxhash(uid)	on delete set null on update cascade) Engine=InnoDB;")
-	defer checkedExec(t, conn, "drop table t7_fk, t7_xxhash")
-
-	query := "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as name, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk using (constraint_schema, constraint_name) where fk.referenced_column_name is not null and fk.table_schema = database() and fk.table_name = 't7_fk' and rc.constraint_schema = database() and rc.table_name = 't7_fk'"
-	assertMatches(t, conn, query, `[[VARCHAR("t7_xxhash") VARCHAR("uid") VARCHAR("t7_uid") VARCHAR("t7_fk_ibfk_1") VARCHAR("CASCADE") VARCHAR("SET NULL")]]`)
-}
-
-func TestConnectWithSystemSchema(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	ctx := context.Background()
-	for _, dbname := range []string{"information_schema", "mysql", "performance_schema", "sys"} {
-		connParams := vtParams
-		connParams.DbName = dbname
-		conn, err := mysql.Connect(ctx, &connParams)
-		require.NoError(t, err)
-		checkedExec(t, conn, `select @@max_allowed_packet from dual`)
-		conn.Close()
-	}
-}
-
-func TestUseSystemSchema(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-	for _, dbname := range []string{"information_schema", "mysql", "performance_schema", "sys"} {
-		checkedExec(t, conn, fmt.Sprintf("use %s", dbname))
-		checkedExec(t, conn, `select @@max_allowed_packet from dual`)
-	}
-}
-
-func TestSystemSchemaQueryWithoutQualifier(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	queryWithQualifier := fmt.Sprintf("select t.table_schema,t.table_name,c.column_name,c.column_type "+
-		"from information_schema.tables t "+
-		"join information_schema.columns c "+
-		"on c.table_schema = t.table_schema and c.table_name = t.table_name "+
-		"where t.table_schema = '%s' and c.table_schema = '%s' "+
-		"order by t.table_schema,t.table_name,c.column_name", KeyspaceName, KeyspaceName)
-	qr1 := checkedExec(t, conn, queryWithQualifier)
-
-	checkedExec(t, conn, "use information_schema")
-	queryWithoutQualifier := fmt.Sprintf("select t.table_schema,t.table_name,c.column_name,c.column_type "+
-		"from tables t "+
-		"join columns c "+
-		"on c.table_schema = t.table_schema and c.table_name = t.table_name "+
-		"where t.table_schema = '%s' and c.table_schema = '%s' "+
-		"order by t.table_schema,t.table_name,c.column_name", KeyspaceName, KeyspaceName)
-	qr2 := checkedExec(t, conn, queryWithoutQualifier)
-	require.Equal(t, qr1, qr2)
-
-	connParams := vtParams
-	connParams.DbName = "information_schema"
-	conn2, err := mysql.Connect(ctx, &connParams)
-	require.NoError(t, err)
-	defer conn2.Close()
-
-	qr3 := checkedExec(t, conn2, queryWithoutQualifier)
-	require.Equal(t, qr2, qr3)
-}
-
-func TestMultipleSchemaPredicates(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	query := fmt.Sprintf("select t.table_schema,t.table_name,c.column_name,c.column_type "+
-		"from information_schema.tables t "+
-		"join information_schema.columns c "+
-		"on c.table_schema = t.table_schema and c.table_name = t.table_name "+
-		"where t.table_schema = '%s' and c.table_schema = '%s' and c.table_schema = '%s' and c.table_schema = '%s'", KeyspaceName, KeyspaceName, KeyspaceName, KeyspaceName)
-	qr1 := checkedExec(t, conn, query)
-	require.EqualValues(t, 4, len(qr1.Fields))
-
-	// test a query with two keyspace names
-	query = fmt.Sprintf("select t.table_schema,t.table_name,c.column_name,c.column_type "+
-		"from information_schema.tables t "+
-		"join information_schema.columns c "+
-		"on c.table_schema = t.table_schema and c.table_name = t.table_name "+
-		"where t.table_schema = '%s' and c.table_schema = '%s' and c.table_schema = '%s'", KeyspaceName, KeyspaceName, "a")
-	_, err = conn.ExecuteFetch(query, 1000, true)
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "specifying two different database in the query is not supported")
-}
diff --git a/go/test/endtoend/vtgate/mysql80/misc_test.go b/go/test/endtoend/vtgate/mysql80/misc_test.go
index c0e50459ea..0f614535dc 100644
--- a/go/test/endtoend/vtgate/mysql80/misc_test.go
+++ b/go/test/endtoend/vtgate/mysql80/misc_test.go
@@ -18,11 +18,8 @@ package vtgate
 
 import (
 	"context"
-	"fmt"
 	"testing"
 
-	"github.com/google/go-cmp/cmp"
-
 	"vitess.io/vitess/go/test/endtoend/cluster"
 
 	"github.com/stretchr/testify/require"
@@ -74,35 +71,6 @@ ts12 TIMESTAMP DEFAULT LOCALTIME()
 	exec(t, conn, "drop table function_default")
 }
 
-// TestCheckConstraint test check constraints on CREATE TABLE
-// This feature is supported from MySQL 8.0.16 and MariaDB 10.2.1.
-func TestCheckConstraint(t *testing.T) {
-	conn, err := mysql.Connect(context.Background(), &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	query := `CREATE TABLE t7 (CHECK (c1 <> c2), c1 INT CHECK (c1 > 10), c2 INT CONSTRAINT c2_positive CHECK (c2 > 0), c3 INT CHECK (c3 < 100), CONSTRAINT c1_nonzero CHECK (c1 <> 0), CHECK (c1 > c3));`
-	exec(t, conn, query)
-
-	checkQuery := `SELECT CONSTRAINT_NAME FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS WHERE TABLE_NAME = 't7' order by CONSTRAINT_NAME;`
-	expected := `[[VARCHAR("c1_nonzero")] [VARCHAR("c2_positive")] [VARCHAR("t7_chk_1")] [VARCHAR("t7_chk_2")] [VARCHAR("t7_chk_3")] [VARCHAR("t7_chk_4")]]`
-
-	assertMatches(t, conn, checkQuery, expected)
-
-	cleanup := `DROP TABLE t7`
-	exec(t, conn, cleanup)
-}
-
-func assertMatches(t *testing.T, conn *mysql.Conn, query, expected string) {
-	t.Helper()
-	qr := exec(t, conn, query)
-	got := fmt.Sprintf("%v", qr.Rows)
-	diff := cmp.Diff(expected, got)
-	if diff != "" {
-		t.Errorf("Query: %s (-want +got):\n%s", query, diff)
-	}
-}
-
 func exec(t *testing.T, conn *mysql.Conn, query string) *sqltypes.Result {
 	t.Helper()
 	qr, err := conn.ExecuteFetch(query, 1000, true)
diff --git a/go/test/endtoend/vtgate/system_schema_test.go b/go/test/endtoend/vtgate/system_schema_test.go
index ee400184c6..ad2fb0f5e9 100644
--- a/go/test/endtoend/vtgate/system_schema_test.go
+++ b/go/test/endtoend/vtgate/system_schema_test.go
@@ -29,6 +29,28 @@ import (
 	"vitess.io/vitess/go/test/endtoend/cluster"
 )
 
+// TestCheckConstraint test check constraints on CREATE TABLE
+// This feature is supported from MySQL 8.0.16 and MariaDB 10.2.1.
+func TestCheckConstraint(t *testing.T) {
+	// Skipping as tests are run against MySQL 5.7
+	t.Skip()
+
+	conn, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	query := `CREATE TABLE t7 (CHECK (c1 <> c2), c1 INT CHECK (c1 > 10), c2 INT CONSTRAINT c2_positive CHECK (c2 > 0), c3 INT CHECK (c3 < 100), CONSTRAINT c1_nonzero CHECK (c1 <> 0), CHECK (c1 > c3));`
+	exec(t, conn, query)
+
+	checkQuery := `SELECT CONSTRAINT_NAME FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS WHERE TABLE_NAME = 't7';`
+	expected := `[[VARCHAR("t7_chk_1")] [VARCHAR("t7_chk_2")] [VARCHAR("c2_positive")] [VARCHAR("t7_chk_3")] [VARCHAR("c1_nonzero")] [VARCHAR("t7_chk_4")]]`
+
+	assertMatches(t, conn, checkQuery, expected)
+
+	cleanup := `DROP TABLE t7`
+	exec(t, conn, cleanup)
+}
+
 func TestDbNameOverride(t *testing.T) {
 	defer cluster.PanicHandler(t)
 	ctx := context.Background()
@@ -147,17 +169,15 @@ func TestSystemSchemaQueryWithoutQualifier(t *testing.T) {
 		"from information_schema.tables t "+
 		"join information_schema.columns c "+
 		"on c.table_schema = t.table_schema and c.table_name = t.table_name "+
-		"where t.table_schema = '%s' and c.table_schema = '%s' "+
-		"order by t.table_schema,t.table_name,c.column_name", KeyspaceName, KeyspaceName)
+		"where t.table_schema = '%s' and c.table_schema = '%s'", KeyspaceName, KeyspaceName)
 	qr1 := exec(t, conn, queryWithQualifier)
 
-	exec(t, conn, "use information_schema")
 	queryWithoutQualifier := fmt.Sprintf("select t.table_schema,t.table_name,c.column_name,c.column_type "+
 		"from tables t "+
 		"join columns c "+
 		"on c.table_schema = t.table_schema and c.table_name = t.table_name "+
-		"where t.table_schema = '%s' and c.table_schema = '%s' "+
-		"order by t.table_schema,t.table_name,c.column_name", KeyspaceName, KeyspaceName)
+		"where t.table_schema = '%s' and c.table_schema = '%s'", KeyspaceName, KeyspaceName)
+	exec(t, conn, "use information_schema")
 	qr2 := exec(t, conn, queryWithoutQualifier)
 	require.Equal(t, qr1, qr2)
 
diff --git a/go/vt/sqlparser/parse_test.go b/go/vt/sqlparser/parse_test.go
index 1c24729bc5..812d937d32 100644
--- a/go/vt/sqlparser/parse_test.go
+++ b/go/vt/sqlparser/parse_test.go
@@ -1178,9 +1178,6 @@ var (
 	}, {
 		input:  "create table function_default3 (x bool DEFAULT (true AND false));",
 		output: "create table function_default3 (\n\tx bool default (true and false)\n)",
-	}, {
-		input:  "create table function_default (x bool DEFAULT true);",
-		output: "create table function_default (\n\tx bool default true\n)",
 	}, {
 		input:  "create table a (\n\t`a` int\n)",
 		output: "create table a (\n\ta int\n)",
diff --git a/go/vt/vtgate/engine/fuzz_flaky_test.go b/go/vt/vtgate/engine/fuzz_flaky_test.go
index 1eb93bf3a0..538875e9a7 100644
--- a/go/vt/vtgate/engine/fuzz_flaky_test.go
+++ b/go/vt/vtgate/engine/fuzz_flaky_test.go
@@ -220,7 +220,7 @@ func execRouteSelectDBA(query, field, tablename, schema, shards string) {
 		Query:               query,
 		FieldQuery:          field,
 		SysTableTableSchema: stringToExpr(schema),
-		SysTableTableName:   map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte(tablename))},
+		SysTableTableName:   stringToExpr(tablename),
 	}
 	vc := &loggingVCursor{
 		shards:  []string{shards},
diff --git a/go/vt/vtgate/engine/route_test.go b/go/vt/vtgate/engine/route_test.go
index d7cad32ba6..aee0abf936 100644
--- a/go/vt/vtgate/engine/route_test.go
+++ b/go/vt/vtgate/engine/route_test.go
@@ -88,57 +88,54 @@ func TestSelectInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T)
 	}
 
 	type testCase struct {
-		tableSchema []string
-		tableName   map[string]evalengine.Expr
-		testName    string
-		expectedLog []string
-		routed      bool
+		tableSchema, tableName []string
+		testName               string
+		expectedLog            []string
+		routed                 bool
 	}
 	tests := []testCase{{
 		testName:    "both schema and table predicates - routed table",
 		tableSchema: []string{"schema"},
-		tableName:   map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("table"))},
+		tableName:   []string{"table"},
 		routed:      true,
 		expectedLog: []string{
 			"FindTable(`schema`.`table`)",
 			"ResolveDestinations routedKeyspace [] Destinations:DestinationAnyShard()",
-			"ExecuteMultiShard routedKeyspace.1: dummy_select {__replacevtschemaname: type:INT64 value:\"1\" table_name: type:VARBINARY value:\"routedTable\"} false false"},
+			"ExecuteMultiShard routedKeyspace.1: dummy_select {__replacevtschemaname: type:INT64 value:\"1\" __vttablename: type:VARBINARY value:\"routedTable\"} false false"},
 	}, {
 		testName:    "both schema and table predicates - not routed",
 		tableSchema: []string{"schema"},
-		tableName:   map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("table"))},
+		tableName:   []string{"table"},
 		routed:      false,
 		expectedLog: []string{
 			"FindTable(`schema`.`table`)",
 			"ResolveDestinations schema [] Destinations:DestinationAnyShard()",
-			"ExecuteMultiShard schema.1: dummy_select {__replacevtschemaname: type:INT64 value:\"1\" table_name: type:VARBINARY value:\"table\"} false false"},
+			"ExecuteMultiShard schema.1: dummy_select {__replacevtschemaname: type:INT64 value:\"1\" __vttablename: type:VARBINARY value:\"table\"} false false"},
 	}, {
 		testName:    "multiple schema and table predicates",
 		tableSchema: []string{"schema", "schema", "schema"},
-		tableName:   map[string]evalengine.Expr{"t1": evalengine.NewLiteralString([]byte("table")), "t2": evalengine.NewLiteralString([]byte("table")), "t3": evalengine.NewLiteralString([]byte("table"))},
+		tableName:   []string{"table", "table", "table"},
 		routed:      false,
 		expectedLog: []string{
-			"FindTable(`schema`.`table`)",
-			"FindTable(`schema`.`table`)",
 			"FindTable(`schema`.`table`)",
 			"ResolveDestinations schema [] Destinations:DestinationAnyShard()",
-			"ExecuteMultiShard schema.1: dummy_select {__replacevtschemaname: type:INT64 value:\"1\" t1: type:VARBINARY value:\"table\" t2: type:VARBINARY value:\"table\" t3: type:VARBINARY value:\"table\"} false false"},
+			"ExecuteMultiShard schema.1: dummy_select {__replacevtschemaname: type:INT64 value:\"1\" __vttablename: type:VARBINARY value:\"table\"} false false"},
 	}, {
 		testName:  "table name predicate - routed table",
-		tableName: map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("tableName"))},
+		tableName: []string{"tableName"},
 		routed:    true,
 		expectedLog: []string{
 			"FindTable(tableName)",
 			"ResolveDestinations routedKeyspace [] Destinations:DestinationAnyShard()",
-			"ExecuteMultiShard routedKeyspace.1: dummy_select {table_name: type:VARBINARY value:\"routedTable\"} false false"},
+			"ExecuteMultiShard routedKeyspace.1: dummy_select {__vttablename: type:VARBINARY value:\"routedTable\"} false false"},
 	}, {
 		testName:  "table name predicate - not routed",
-		tableName: map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("tableName"))},
+		tableName: []string{"tableName"},
 		routed:    false,
 		expectedLog: []string{
 			"FindTable(tableName)",
 			"ResolveDestinations ks [] Destinations:DestinationAnyShard()",
-			"ExecuteMultiShard ks.1: dummy_select {table_name: type:VARBINARY value:\"tableName\"} false false"},
+			"ExecuteMultiShard ks.1: dummy_select {__vttablename: type:VARBINARY value:\"tableName\"} false false"},
 	}, {
 		testName:    "schema predicate",
 		tableSchema: []string{"myKeyspace"},
@@ -157,6 +154,7 @@ func TestSelectInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T)
 			"ResolveDestinations ks [] Destinations:DestinationAnyShard()",
 			"ExecuteMultiShard ks.1: dummy_select {} false false"},
 	}}
+
 	for _, tc := range tests {
 		t.Run(tc.testName, func(t *testing.T) {
 			sel := &Route{
@@ -168,7 +166,7 @@ func TestSelectInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T)
 				Query:               "dummy_select",
 				FieldQuery:          "dummy_select_field",
 				SysTableTableSchema: stringListToExprList(tc.tableSchema),
-				SysTableTableName:   tc.tableName,
+				SysTableTableName:   stringListToExprList(tc.tableName),
 			}
 			vc := &loggingVCursor{
 				shards:  []string{"1"},
diff --git a/go/vt/vtgate/executor_select_test.go b/go/vt/vtgate/executor_select_test.go
index 413cbb4eb0..05c1ba5af7 100644
--- a/go/vt/vtgate/executor_select_test.go
+++ b/go/vt/vtgate/executor_select_test.go
@@ -74,16 +74,16 @@ func TestSelectDBA(t *testing.T) {
 	utils.MustMatch(t, wantQueries, sbc1.Queries)
 
 	sbc1.Queries = nil
-	query = "SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES ist WHERE ist.table_schema = 'performance_schema' AND ist.table_name = 'foo'"
+	query = "SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE table_schema = 'performance_schema' AND table_name = 'foo'"
 	_, err = executor.Execute(context.Background(), "TestSelectDBA",
 		NewSafeSession(&vtgatepb.Session{TargetString: "TestExecutor"}),
 		query, map[string]*querypb.BindVariable{},
 	)
 	require.NoError(t, err)
-	wantQueries = []*querypb.BoundQuery{{Sql: "select COUNT(*) from INFORMATION_SCHEMA.`TABLES` as ist where ist.table_schema = :__vtschemaname and ist.table_name = :ist_table_name",
+	wantQueries = []*querypb.BoundQuery{{Sql: "select COUNT(*) from INFORMATION_SCHEMA.`TABLES` where table_schema = :__vtschemaname and table_name = :__vttablename",
 		BindVariables: map[string]*querypb.BindVariable{
 			"__vtschemaname": sqltypes.StringBindVariable("performance_schema"),
-			"ist_table_name": sqltypes.StringBindVariable("foo"),
+			"__vttablename":  sqltypes.StringBindVariable("foo"),
 		}}}
 	utils.MustMatch(t, wantQueries, sbc1.Queries)
 
@@ -94,10 +94,10 @@ func TestSelectDBA(t *testing.T) {
 		query, map[string]*querypb.BindVariable{},
 	)
 	require.NoError(t, err)
-	wantQueries = []*querypb.BoundQuery{{Sql: "select 1 from information_schema.table_constraints where constraint_schema = :__vtschemaname and table_name = :table_name",
+	wantQueries = []*querypb.BoundQuery{{Sql: "select 1 from information_schema.table_constraints where constraint_schema = :__vtschemaname and table_name = :__vttablename",
 		BindVariables: map[string]*querypb.BindVariable{
 			"__vtschemaname": sqltypes.StringBindVariable("vt_ks"),
-			"table_name":     sqltypes.StringBindVariable("user"),
+			"__vttablename":  sqltypes.StringBindVariable("user"),
 		}}}
 	utils.MustMatch(t, wantQueries, sbc1.Queries)
 
@@ -115,87 +115,6 @@ func TestSelectDBA(t *testing.T) {
 	utils.MustMatch(t, wantQueries, sbc1.Queries)
 }
 
-func TestGen4SelectDBA(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
-	executor.normalize = true
-	*plannerVersion = "gen4"
-	defer func() {
-		// change it back to v3
-		*plannerVersion = "v3"
-	}()
-
-	query := "select * from INFORMATION_SCHEMA.foo"
-	_, err := executor.Execute(context.Background(), "TestSelectDBA",
-		NewSafeSession(&vtgatepb.Session{TargetString: "TestExecutor"}),
-		query, map[string]*querypb.BindVariable{},
-	)
-	require.NoError(t, err)
-	wantQueries := []*querypb.BoundQuery{{Sql: query, BindVariables: map[string]*querypb.BindVariable{}}}
-	utils.MustMatch(t, wantQueries, sbc1.Queries)
-
-	sbc1.Queries = nil
-	query = "SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES ist WHERE ist.table_schema = 'performance_schema' AND ist.table_name = 'foo'"
-	_, err = executor.Execute(context.Background(), "TestSelectDBA",
-		NewSafeSession(&vtgatepb.Session{TargetString: "TestExecutor"}),
-		query, map[string]*querypb.BindVariable{},
-	)
-	require.NoError(t, err)
-	wantQueries = []*querypb.BoundQuery{{Sql: "select COUNT(*) from INFORMATION_SCHEMA.`TABLES` as ist where ist.table_schema = :__vtschemaname and ist.table_name = :ist_table_name",
-		BindVariables: map[string]*querypb.BindVariable{
-			"vtg1":           sqltypes.StringBindVariable("performance_schema"),
-			"vtg2":           sqltypes.StringBindVariable("foo"),
-			"__vtschemaname": sqltypes.StringBindVariable("performance_schema"),
-			"ist_table_name": sqltypes.StringBindVariable("foo"),
-		}}}
-	utils.MustMatch(t, wantQueries, sbc1.Queries)
-
-	sbc1.Queries = nil
-	query = "select 1 from information_schema.table_constraints where constraint_schema = 'vt_ks' and table_name = 'user'"
-	_, err = executor.Execute(context.Background(), "TestSelectDBA",
-		NewSafeSession(&vtgatepb.Session{TargetString: "TestExecutor"}),
-		query, map[string]*querypb.BindVariable{},
-	)
-	require.NoError(t, err)
-	wantQueries = []*querypb.BoundQuery{{Sql: "select :vtg1 from information_schema.table_constraints where constraint_schema = :__vtschemaname and table_name = :table_name",
-		BindVariables: map[string]*querypb.BindVariable{
-			"vtg1":           sqltypes.Int64BindVariable(1),
-			"vtg2":           sqltypes.StringBindVariable("vt_ks"),
-			"vtg3":           sqltypes.StringBindVariable("user"),
-			"__vtschemaname": sqltypes.StringBindVariable("vt_ks"),
-			"table_name":     sqltypes.StringBindVariable("user"),
-		}}}
-	utils.MustMatch(t, wantQueries, sbc1.Queries)
-
-	sbc1.Queries = nil
-	query = "select 1 from information_schema.table_constraints where constraint_schema = 'vt_ks'"
-	_, err = executor.Execute(context.Background(), "TestSelectDBA",
-		NewSafeSession(&vtgatepb.Session{TargetString: "TestExecutor"}),
-		query, map[string]*querypb.BindVariable{},
-	)
-	require.NoError(t, err)
-	wantQueries = []*querypb.BoundQuery{{Sql: "select :vtg1 from information_schema.table_constraints where constraint_schema = :__vtschemaname",
-		BindVariables: map[string]*querypb.BindVariable{
-			"vtg1":           sqltypes.Int64BindVariable(1),
-			"vtg2":           sqltypes.StringBindVariable("vt_ks"),
-			"__vtschemaname": sqltypes.StringBindVariable("vt_ks"),
-		}}}
-	utils.MustMatch(t, wantQueries, sbc1.Queries)
-
-	sbc1.Queries = nil
-	query = "select t.table_schema,t.table_name,c.column_name,c.column_type from tables t join columns c on c.table_schema = t.table_schema and c.table_name = t.table_name where t.table_schema = 'TestExecutor' and c.table_schema = 'TestExecutor' order by t.table_schema,t.table_name,c.column_name"
-	_, err = executor.Execute(context.Background(), "TestSelectDBA",
-		NewSafeSession(&vtgatepb.Session{TargetString: "information_schema"}),
-		query, map[string]*querypb.BindVariable{},
-	)
-	require.NoError(t, err)
-	wantQueries = []*querypb.BoundQuery{{Sql: "select t.table_schema, t.table_name, c.column_name, c.column_type from information_schema.`tables` as t, information_schema.`columns` as c where t.table_schema = :__vtschemaname and c.table_schema = :__vtschemaname and c.table_schema = t.table_schema and c.table_name = t.table_name order by t.table_schema asc, t.table_name asc, c.column_name asc",
-		BindVariables: map[string]*querypb.BindVariable{
-			"vtg1":                  sqltypes.StringBindVariable("TestExecutor"),
-			"__replacevtschemaname": sqltypes.Int64BindVariable(1),
-		}}}
-	utils.MustMatch(t, wantQueries, sbc1.Queries)
-}
-
 func TestUnsharded(t *testing.T) {
 	executor, _, _, sbclookup := createLegacyExecutorEnv()
 
diff --git a/go/vt/vtgate/planbuilder/plan_test.go b/go/vt/vtgate/planbuilder/plan_test.go
index f3ca82587e..4778728ea9 100644
--- a/go/vt/vtgate/planbuilder/plan_test.go
+++ b/go/vt/vtgate/planbuilder/plan_test.go
@@ -195,7 +195,6 @@ func TestPlan(t *testing.T) {
 	testFile(t, "flush_cases_no_default_keyspace.txt", testOutputTempDir, vschemaWrapper, false)
 	testFile(t, "show_cases_no_default_keyspace.txt", testOutputTempDir, vschemaWrapper, false)
 	testFile(t, "stream_cases.txt", testOutputTempDir, vschemaWrapper, false)
-	testFile(t, "systemtables_cases.txt", testOutputTempDir, vschemaWrapper, false)
 }
 
 func TestSysVarSetDisabled(t *testing.T) {
@@ -287,7 +286,7 @@ func TestWithSystemSchemaAsDefaultKeyspace(t *testing.T) {
 	defer os.RemoveAll(testOutputTempDir)
 	vschema := &vschemaWrapper{
 		v:          loadSchema(t, "schema_test.json"),
-		keyspace:   &vindexes.Keyspace{Name: "information_schema"},
+		keyspace:   &vindexes.Keyspace{Name: "mysql"},
 		tabletType: topodatapb.TabletType_MASTER,
 	}
 
@@ -469,10 +468,6 @@ func escapeNewLines(in string) string {
 }
 
 func testFile(t *testing.T, filename, tempDir string, vschema *vschemaWrapper, checkGen4equalPlan bool) {
-	ksName := ""
-	if vschema.keyspace != nil {
-		ksName = vschema.keyspace.Name
-	}
 	var checkAllTests = false
 	t.Run(filename, func(t *testing.T) {
 		expected := &strings.Builder{}
@@ -481,7 +476,7 @@ func testFile(t *testing.T, filename, tempDir string, vschema *vschemaWrapper, c
 		for tcase := range iterateExecFile(filename) {
 			t.Run(fmt.Sprintf("%d V3: %s", tcase.lineno, tcase.comments), func(t *testing.T) {
 				vschema.version = V3
-				plan, err := TestBuilder(tcase.input, vschema, ksName)
+				plan, err := TestBuilder(tcase.input, vschema)
 				out := getPlanOrErrorOutput(err, plan)
 
 				if out != tcase.output {
@@ -558,11 +553,7 @@ func getPlanOutput(tcase testCase, vschema *vschemaWrapper) (out string, err err
 			out = fmt.Sprintf("panicked: %v\n%s", r, string(debug.Stack()))
 		}
 	}()
-	ksName := ""
-	if vschema.keyspace != nil {
-		ksName = vschema.keyspace.Name
-	}
-	plan, err := TestBuilder(tcase.input, vschema, ksName)
+	plan, err := TestBuilder(tcase.input, vschema)
 	out = getPlanOrErrorOutput(err, plan)
 	return out, err
 }
@@ -747,16 +738,12 @@ func BenchmarkSelectVsDML(b *testing.B) {
 }
 
 func benchmarkPlanner(b *testing.B, version PlannerVersion, testCases []testCase, vschema *vschemaWrapper) {
-	ksName := ""
-	if vschema.keyspace != nil {
-		ksName = vschema.keyspace.Name
-	}
 	b.ReportAllocs()
 	for n := 0; n < b.N; n++ {
 		for _, tcase := range testCases {
 			if tcase.output2ndPlanner != "" {
 				vschema.version = version
-				_, _ = TestBuilder(tcase.input, vschema, ksName)
+				_, _ = TestBuilder(tcase.input, vschema)
 			}
 		}
 	}
diff --git a/go/vt/vtgate/planbuilder/route_planning_test.go b/go/vt/vtgate/planbuilder/route_planning_test.go
index e392ae2503..75b656c88c 100644
--- a/go/vt/vtgate/planbuilder/route_planning_test.go
+++ b/go/vt/vtgate/planbuilder/route_planning_test.go
@@ -82,11 +82,11 @@ func TestMergeJoins(t *testing.T) {
 	}, {
 		l:        selectDBA(1, ks),
 		r:        selectDBA(2, ks2),
-		expected: selectDBA(1|2, ks),
+		expected: nil,
 	}, {
 		l:        selectDBA(2, ks),
 		r:        selectDBA(1, ks2),
-		expected: selectDBA(1|2, ks),
+		expected: nil,
 	}, {
 		l:        unsharded(1, ks),
 		r:        selectDBA(2, ks),
diff --git a/go/vt/vtgate/planbuilder/testdata/filter_cases.txt b/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
index 7ad39a0391..08422d1fc7 100644
--- a/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
@@ -1172,6 +1172,24 @@ Gen4 plan same as above
 }
 Gen4 plan same as above
 
+# subquery of information_schema with itself
+"select * from information_schema.a where id in (select * from information_schema.b)"
+{
+  "QueryType": "SELECT",
+  "Original": "select * from information_schema.a where id in (select * from information_schema.b)",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from information_schema.a where 1 != 1",
+    "Query": "select * from information_schema.a where id in (select * from information_schema.b)",
+    "Table": "information_schema.a"
+  }
+}
+
 # subquery
 "select u.m from user_extra join user u where u.id in (select m2 from user where user.id = u.id and user_extra.col = user.col) and u.id in (user_extra.col, 1)"
 {
@@ -1895,6 +1913,102 @@ Gen4 plan same as above
 }
 Gen4 plan same as above
 
+# query trying to query two different keyspaces at the same time
+"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'user' AND TABLE_SCHEMA = 'main'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'user' AND TABLE_SCHEMA = 'main'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
+    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = :__vtschemaname and TABLE_SCHEMA = :__vtschemaname",
+    "SysTableTableSchema": "[VARBINARY(\"user\"), VARBINARY(\"main\")]",
+    "Table": "INFORMATION_SCHEMA.`TABLES`"
+  }
+}
+
+# information_schema query using database() func
+"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = database()"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = database()",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
+    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = database()",
+    "Table": "INFORMATION_SCHEMA.`TABLES`"
+  }
+}
+Gen4 plan same as above
+
+# table_schema predicate the wrong way around
+"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE 'ks' = TABLE_SCHEMA"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE 'ks' = TABLE_SCHEMA",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
+    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = :__vtschemaname",
+    "SysTableTableSchema": "[VARBINARY(\"ks\")]",
+    "Table": "INFORMATION_SCHEMA.`TABLES`"
+  }
+}
+
+# table_name predicate against a routed table
+"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'ks' AND TABLE_NAME = 'route1'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'ks' AND TABLE_NAME = 'route1'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
+    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = :__vtschemaname and TABLE_NAME = :__vttablename",
+    "SysTableTableName": "[VARBINARY(\"route1\")]",
+    "SysTableTableSchema": "[VARBINARY(\"ks\")]",
+    "Table": "INFORMATION_SCHEMA.`TABLES`"
+  }
+}
+
+# information_schema query with additional predicates
+"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'ks' and other_column = 42"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'ks' and other_column = 42",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
+    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = :__vtschemaname and other_column = 42",
+    "SysTableTableSchema": "[VARBINARY(\"ks\")]",
+    "Table": "INFORMATION_SCHEMA.`TABLES`"
+  }
+}
+
 # pullout sq after pullout sq
 "select id from user where not id in (select user_extra.col from user_extra where user_extra.user_id = 42) and id in (select user_extra.col from user_extra where user_extra.user_id = 411)"
 {
@@ -1959,6 +2073,25 @@ Gen4 plan same as above
   }
 }
 
+# able to isolate table_schema value even when hidden inside of ORs
+"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE (TABLE_SCHEMA = 'ks' and other_column = 42) OR (TABLE_SCHEMA = 'ks' and foobar = 'value')"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE (TABLE_SCHEMA = 'ks' and other_column = 42) OR (TABLE_SCHEMA = 'ks' and foobar = 'value')",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
+    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = :__vtschemaname and (other_column = 42 or TABLE_SCHEMA = 'ks') and (other_column = 42 or foobar = 'value')",
+    "SysTableTableSchema": "[VARBINARY(\"ks\")]",
+    "Table": "INFORMATION_SCHEMA.`TABLES`"
+  }
+}
+
 # solving LIKE query with a CFC prefix vindex
 "select c2 from cfc_vindex_col where c1 like 'A%'"
 {
diff --git a/go/vt/vtgate/planbuilder/testdata/from_cases.txt b/go/vt/vtgate/planbuilder/testdata/from_cases.txt
index 57a2fd9b4d..83539e3704 100644
--- a/go/vt/vtgate/planbuilder/testdata/from_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/from_cases.txt
@@ -74,6 +74,25 @@ Gen4 plan same as above
 }
 Gen4 plan same as above
 
+# Single information_schema query
+"select col from information_schema.foo"
+{
+  "QueryType": "SELECT",
+  "Original": "select col from information_schema.foo",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select col from information_schema.foo where 1 != 1",
+    "Query": "select col from information_schema.foo",
+    "Table": "information_schema.foo"
+  }
+}
+Gen4 plan same as above
+
 # Multi-table unsharded
 "select m1.col from unsharded as m1 join unsharded as m2"
 {
@@ -302,6 +321,25 @@ Gen4 plan same as above
 }
 Gen4 plan same as above
 
+# ',' join information_schema
+"select a.id,b.id from information_schema.a as a, information_schema.b as b"
+{
+  "QueryType": "SELECT",
+  "Original": "select a.id,b.id from information_schema.a as a, information_schema.b as b",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select a.id, b.id from information_schema.a as a, information_schema.b as b where 1 != 1",
+    "Query": "select a.id, b.id from information_schema.a as a, information_schema.b as b",
+    "Table": "information_schema.a, information_schema.b"
+  }
+}
+Gen4 plan same as above
+
 # ',' 3-way join unsharded
 "select u1.a, u2.a from unsharded u1, unsharded u2, unsharded u3"
 {
@@ -1795,6 +1833,44 @@ Gen4 plan same as above
   }
 }
 
+# access to unqualified column names in information_schema
+"select * from information_schema.a where b=10"
+{
+  "QueryType": "SELECT",
+  "Original": "select * from information_schema.a where b=10",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from information_schema.a where 1 != 1",
+    "Query": "select * from information_schema.a where b = 10",
+    "Table": "information_schema.a"
+  }
+}
+Gen4 plan same as above
+
+# access to qualified column names in information_schema
+"select * from information_schema.a where information_schema.a.b=10"
+{
+  "QueryType": "SELECT",
+  "Original": "select * from information_schema.a where information_schema.a.b=10",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from information_schema.a where 1 != 1",
+    "Query": "select * from information_schema.a where information_schema.a.b = 10",
+    "Table": "information_schema.a"
+  }
+}
+Gen4 plan same as above
+
 # join of information_schema with normal table
 "select unsharded.foo from information_schema.a join unsharded"
 {
@@ -2715,6 +2791,293 @@ Gen4 plan same as above
 "table t not found"
 Gen4 plan same as above
 
+# information schema query that uses table_schema
+"select column_name from information_schema.columns where table_schema = (select schema())"
+{
+  "QueryType": "SELECT",
+  "Original": "select column_name from information_schema.columns where table_schema = (select schema())",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select column_name from information_schema.`columns` where 1 != 1",
+    "Query": "select column_name from information_schema.`columns` where table_schema = schema()",
+    "Table": "information_schema.`columns`"
+  }
+}
+Gen4 plan same as above
+
+# information schema join
+"select * from information_schema.a join information_schema.b"
+{
+  "QueryType": "SELECT",
+  "Original": "select * from information_schema.a join information_schema.b",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from information_schema.a join information_schema.b where 1 != 1",
+    "Query": "select * from information_schema.a join information_schema.b",
+    "Table": "information_schema.a, information_schema.b"
+  }
+}
+{
+  "QueryType": "SELECT",
+  "Original": "select * from information_schema.a join information_schema.b",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from information_schema.a, information_schema.b where 1 != 1",
+    "Query": "select * from information_schema.a, information_schema.b",
+    "Table": "information_schema.a, information_schema.b"
+  }
+}
+
+# rails query
+"select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as name, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk using (constraint_schema, constraint_name) where fk.referenced_column_name is not null and fk.table_schema = database() and fk.table_name = ':vtg1' and rc.constraint_schema = database() and rc.table_name = ':vtg1'"
+{
+  "QueryType": "SELECT",
+  "Original": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as name, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk using (constraint_schema, constraint_name) where fk.referenced_column_name is not null and fk.table_schema = database() and fk.table_name = ':vtg1' and rc.constraint_schema = database() and rc.table_name = ':vtg1'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk on rc.constraint_schema = fk.constraint_schema and rc.constraint_name = fk.constraint_name where 1 != 1",
+    "Query": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk on rc.constraint_schema = fk.constraint_schema and rc.constraint_name = fk.constraint_name where fk.referenced_column_name is not null and fk.table_schema = database() and fk.table_name = :__vttablename and rc.constraint_schema = database() and rc.table_name = :__vttablename",
+    "SysTableTableName": "[VARBINARY(\":vtg1\"), VARBINARY(\":vtg1\")]",
+    "Table": "information_schema.referential_constraints, information_schema.key_column_usage"
+  }
+}
+
+#rails_query 2
+"SELECT * FROM information_schema.schemata WHERE schema_name = 'user'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT * FROM information_schema.schemata WHERE schema_name = 'user'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from information_schema.schemata where 1 != 1",
+    "Query": "select * from information_schema.schemata where schema_name = :__vtschemaname",
+    "SysTableTableSchema": "[VARBINARY(\"user\")]",
+    "Table": "information_schema.schemata"
+  }
+}
+
+#rails_query 3
+"SELECT table_comment FROM information_schema.tables WHERE table_schema = 'schema_name' AND table_name = 'table_name'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT table_comment FROM information_schema.tables WHERE table_schema = 'schema_name' AND table_name = 'table_name'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select table_comment from information_schema.`tables` where 1 != 1",
+    "Query": "select table_comment from information_schema.`tables` where table_schema = :__vtschemaname and table_name = :__vttablename",
+    "SysTableTableName": "[VARBINARY(\"table_name\")]",
+    "SysTableTableSchema": "[VARBINARY(\"schema_name\")]",
+    "Table": "information_schema.`tables`"
+  }
+}
+
+#rails_query 4
+"SELECT fk.referenced_table_name AS 'to_table', fk.referenced_column_name AS 'primary_key',fk.column_name AS 'column',fk.constraint_name AS 'name',rc.update_rule AS 'on_update',rc.delete_rule AS 'on_delete' FROM information_schema.referential_constraints rc JOIN information_schema.key_column_usage fk USING (constraint_schema, constraint_name) WHERE fk.referenced_column_name IS NOT NULL AND fk.table_schema = 'table_schema' AND fk.table_name = 'table_name' AND rc.constraint_schema = 'table_schema' AND rc.table_name = 'table_name'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT fk.referenced_table_name AS 'to_table', fk.referenced_column_name AS 'primary_key',fk.column_name AS 'column',fk.constraint_name AS 'name',rc.update_rule AS 'on_update',rc.delete_rule AS 'on_delete' FROM information_schema.referential_constraints rc JOIN information_schema.key_column_usage fk USING (constraint_schema, constraint_name) WHERE fk.referenced_column_name IS NOT NULL AND fk.table_schema = 'table_schema' AND fk.table_name = 'table_name' AND rc.constraint_schema = 'table_schema' AND rc.table_name = 'table_name'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk on rc.constraint_schema = fk.constraint_schema and rc.constraint_name = fk.constraint_name where 1 != 1",
+    "Query": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk on rc.constraint_schema = fk.constraint_schema and rc.constraint_name = fk.constraint_name where fk.referenced_column_name is not null and fk.table_schema = :__vtschemaname and fk.table_name = :__vttablename and rc.constraint_schema = :__vtschemaname and rc.table_name = :__vttablename",
+    "SysTableTableName": "[VARBINARY(\"table_name\"), VARBINARY(\"table_name\")]",
+    "SysTableTableSchema": "[VARBINARY(\"table_schema\"), VARBINARY(\"table_schema\")]",
+    "Table": "information_schema.referential_constraints, information_schema.key_column_usage"
+  }
+}
+
+#rails_query 5
+"SELECT cc.constraint_name AS 'name', cc.check_clause AS 'expression' FROM information_schema.check_constraints cc JOIN information_schema.table_constraints tc USING (constraint_schema, constraint_name) WHERE tc.table_schema = 'table_schema' AND tc.table_name = 'table_name' AND cc.constraint_schema = 'constraint_schema'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT cc.constraint_name AS 'name', cc.check_clause AS 'expression' FROM information_schema.check_constraints cc JOIN information_schema.table_constraints tc USING (constraint_schema, constraint_name) WHERE tc.table_schema = 'table_schema' AND tc.table_name = 'table_name' AND cc.constraint_schema = 'constraint_schema'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select cc.constraint_name as `name`, cc.check_clause as expression from information_schema.check_constraints as cc join information_schema.table_constraints as tc on cc.constraint_schema = tc.constraint_schema and cc.constraint_name = tc.constraint_name where 1 != 1",
+    "Query": "select cc.constraint_name as `name`, cc.check_clause as expression from information_schema.check_constraints as cc join information_schema.table_constraints as tc on cc.constraint_schema = tc.constraint_schema and cc.constraint_name = tc.constraint_name where tc.table_schema = :__vtschemaname and tc.table_name = :__vttablename and cc.constraint_schema = :__vtschemaname",
+    "SysTableTableName": "[VARBINARY(\"table_name\")]",
+    "SysTableTableSchema": "[VARBINARY(\"table_schema\"), VARBINARY(\"constraint_schema\")]",
+    "Table": "information_schema.check_constraints, information_schema.table_constraints"
+  }
+}
+
+#rails_query 6
+"SELECT column_name FROM information_schema.statistics WHERE index_name = 'PRIMARY' AND table_schema = 'table_schema' AND table_name = 'table_name' ORDER BY seq_in_index"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT column_name FROM information_schema.statistics WHERE index_name = 'PRIMARY' AND table_schema = 'table_schema' AND table_name = 'table_name' ORDER BY seq_in_index",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select column_name from information_schema.statistics where 1 != 1",
+    "Query": "select column_name from information_schema.statistics where index_name = 'PRIMARY' and table_schema = :__vtschemaname and table_name = :__vttablename order by seq_in_index asc",
+    "SysTableTableName": "[VARBINARY(\"table_name\")]",
+    "SysTableTableSchema": "[VARBINARY(\"table_schema\")]",
+    "Table": "information_schema.statistics"
+  }
+}
+
+#rails_query 7
+"SELECT generation_expression FROM information_schema.columns WHERE table_schema = 'table_schema' AND table_name = 'table_name' AND column_name = 'column_name'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT generation_expression FROM information_schema.columns WHERE table_schema = 'table_schema' AND table_name = 'table_name' AND column_name = 'column_name'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select generation_expression from information_schema.`columns` where 1 != 1",
+    "Query": "select generation_expression from information_schema.`columns` where table_schema = :__vtschemaname and table_name = :__vttablename and column_name = 'column_name'",
+    "SysTableTableName": "[VARBINARY(\"table_name\")]",
+    "SysTableTableSchema": "[VARBINARY(\"table_schema\")]",
+    "Table": "information_schema.`columns`"
+  }
+}
+
+#rails_query 8
+"SELECT id FROM information_schema.processlist WHERE info LIKE '% FOR UPDATE'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT id FROM information_schema.processlist WHERE info LIKE '% FOR UPDATE'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select id from information_schema.`processlist` where 1 != 1",
+    "Query": "select id from information_schema.`processlist` where info like '% FOR UPDATE'",
+    "Table": "information_schema.`processlist`"
+  }
+}
+Gen4 plan same as above
+
+#rails_query 9
+"SELECT table_name FROM (SELECT * FROM information_schema.tables WHERE table_schema = 'table_schema') _subquery"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT table_name FROM (SELECT * FROM information_schema.tables WHERE table_schema = 'table_schema') _subquery",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select table_name from (select * from information_schema.`tables` where 1 != 1) as _subquery where 1 != 1",
+    "Query": "select table_name from (select * from information_schema.`tables` where table_schema = :__vtschemaname) as _subquery",
+    "SysTableTableSchema": "[VARBINARY(\"table_schema\")]",
+    "Table": "information_schema.`tables`"
+  }
+}
+
+#rails_query 10
+"SELECT table_name FROM (SELECT * FROM information_schema.tables WHERE table_schema = 'table_schema') _subquery WHERE _subquery.table_type = 'table_type' AND _subquery.table_name = 'table_name'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT table_name FROM (SELECT * FROM information_schema.tables WHERE table_schema = 'table_schema') _subquery WHERE _subquery.table_type = 'table_type' AND _subquery.table_name = 'table_name'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select table_name from (select * from information_schema.`tables` where 1 != 1) as _subquery where 1 != 1",
+    "Query": "select table_name from (select * from information_schema.`tables` where table_schema = :__vtschemaname) as _subquery where _subquery.table_type = 'table_type' and _subquery.table_name = :__vttablename",
+    "SysTableTableName": "[VARBINARY(\"table_name\")]",
+    "SysTableTableSchema": "[VARBINARY(\"table_schema\")]",
+    "Table": "information_schema.`tables`"
+  }
+}
+
+# two predicates specifying the database for the same table work if the database is the same
+"SELECT cc.constraint_name AS 'name' FROM information_schema.check_constraints cc  WHERE cc.constraint_schema = 'a' AND cc.table_schema = 'a'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT cc.constraint_name AS 'name' FROM information_schema.check_constraints cc  WHERE cc.constraint_schema = 'a' AND cc.table_schema = 'a'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select cc.constraint_name as `name` from information_schema.check_constraints as cc where 1 != 1",
+    "Query": "select cc.constraint_name as `name` from information_schema.check_constraints as cc where cc.constraint_schema = :__vtschemaname and cc.table_schema = :__vtschemaname",
+    "SysTableTableSchema": "[VARBINARY(\"a\"), VARBINARY(\"a\")]",
+    "Table": "information_schema.check_constraints"
+  }
+}
+
+# system schema in where clause of information_schema query
+"SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE table_schema = 'performance_schema' AND table_name = 'foo'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE table_schema = 'performance_schema' AND table_name = 'foo'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select COUNT(*) from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
+    "Query": "select COUNT(*) from INFORMATION_SCHEMA.`TABLES` where table_schema = :__vtschemaname and table_name = :__vttablename",
+    "SysTableTableName": "[VARBINARY(\"foo\")]",
+    "SysTableTableSchema": "[VARBINARY(\"performance_schema\")]",
+    "Table": "INFORMATION_SCHEMA.`TABLES`"
+  }
+}
+
 # query with parens is planned correctly
 "select m1.col from (unsharded as m1, unsharded as m2)"
 {
diff --git a/go/vt/vtgate/planbuilder/testdata/select_cases.txt b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
index b1c805cc4a..c4992c4b7a 100644
--- a/go/vt/vtgate/planbuilder/testdata/select_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
@@ -1249,6 +1249,25 @@ Gen4 plan same as above
 "select t.col from (select user.id from user join user_extra) as t"
 "symbol t.col not found in table or subquery"
 
+# union of information_schema
+"select * from information_schema.a union select * from information_schema.b"
+{
+  "QueryType": "SELECT",
+  "Original": "select * from information_schema.a union select * from information_schema.b",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select * from information_schema.a where 1 != 1 union select * from information_schema.b where 1 != 1",
+    "Query": "select * from information_schema.a union select * from information_schema.b",
+    "Table": "information_schema.a"
+  }
+}
+Gen4 plan same as above
+
 # union with the same target shard
 "select * from music where user_id = 1 union select * from user where id = 1"
 {
@@ -1754,3 +1773,121 @@ Gen4 plan same as above
   }
 }
 Gen4 plan same as above
+
+# Select from information schema query with two tables that route should be merged
+"SELECT DELETE_RULE, UPDATE_RULE FROM  INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS KCU INNER JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS RC ON KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME WHERE KCU.TABLE_SCHEMA = 'test' AND KCU.TABLE_NAME = 'data_type_table' AND KCU.COLUMN_NAME = 'id' AND KCU.REFERENCED_TABLE_SCHEMA = 'test' AND KCU.CONSTRAINT_NAME = 'data_type_table_id_fkey' ORDER BY KCU.CONSTRAINT_NAME, KCU.COLUMN_NAME"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT DELETE_RULE, UPDATE_RULE FROM  INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS KCU INNER JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS RC ON KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME WHERE KCU.TABLE_SCHEMA = 'test' AND KCU.TABLE_NAME = 'data_type_table' AND KCU.COLUMN_NAME = 'id' AND KCU.REFERENCED_TABLE_SCHEMA = 'test' AND KCU.CONSTRAINT_NAME = 'data_type_table_id_fkey' ORDER BY KCU.CONSTRAINT_NAME, KCU.COLUMN_NAME",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select DELETE_RULE, UPDATE_RULE from INFORMATION_SCHEMA.KEY_COLUMN_USAGE as KCU join INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS as RC on KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME where 1 != 1",
+    "Query": "select DELETE_RULE, UPDATE_RULE from INFORMATION_SCHEMA.KEY_COLUMN_USAGE as KCU join INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS as RC on KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME where KCU.TABLE_SCHEMA = :__vtschemaname and KCU.TABLE_NAME = :__vttablename and KCU.COLUMN_NAME = 'id' and KCU.REFERENCED_TABLE_SCHEMA = 'test' and KCU.CONSTRAINT_NAME = 'data_type_table_id_fkey' order by KCU.CONSTRAINT_NAME asc, KCU.COLUMN_NAME asc",
+    "SysTableTableName": "[VARBINARY(\"data_type_table\")]",
+    "SysTableTableSchema": "[VARBINARY(\"test\")]",
+    "Table": "INFORMATION_SCHEMA.KEY_COLUMN_USAGE, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS"
+  }
+}
+
+# Select from information schema query with three tables such that route for 2 should be merged but not for the last.
+"SELECT KCU.DELETE_RULE, S.UPDATE_RULE FROM  INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS KCU INNER JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS RC ON KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME, INFORMATION_SCHEMA.K AS S WHERE KCU.TABLE_SCHEMA = 'test' AND KCU.TABLE_NAME = 'data_type_table' AND KCU.TABLE_NAME = 'data_type_table' AND S.TABLE_SCHEMA = 'test' AND S.TABLE_NAME = 'sc' ORDER BY KCU.CONSTRAINT_NAME, KCU.COLUMN_NAME"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT KCU.DELETE_RULE, S.UPDATE_RULE FROM  INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS KCU INNER JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS RC ON KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME, INFORMATION_SCHEMA.K AS S WHERE KCU.TABLE_SCHEMA = 'test' AND KCU.TABLE_NAME = 'data_type_table' AND KCU.TABLE_NAME = 'data_type_table' AND S.TABLE_SCHEMA = 'test' AND S.TABLE_NAME = 'sc' ORDER BY KCU.CONSTRAINT_NAME, KCU.COLUMN_NAME",
+  "Instructions": {
+    "OperatorType": "Join",
+    "Variant": "Join",
+    "JoinColumnIndexes": "-1,1",
+    "TableName": "INFORMATION_SCHEMA.KEY_COLUMN_USAGE, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS_INFORMATION_SCHEMA.K",
+    "Inputs": [
+      {
+        "OperatorType": "Route",
+        "Variant": "SelectDBA",
+        "Keyspace": {
+          "Name": "main",
+          "Sharded": false
+        },
+        "FieldQuery": "select KCU.DELETE_RULE from INFORMATION_SCHEMA.KEY_COLUMN_USAGE as KCU join INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS as RC on KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME where 1 != 1",
+        "Query": "select KCU.DELETE_RULE from INFORMATION_SCHEMA.KEY_COLUMN_USAGE as KCU join INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS as RC on KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME where KCU.TABLE_SCHEMA = :__vtschemaname and KCU.TABLE_NAME = :__vttablename and KCU.TABLE_NAME = :__vttablename order by KCU.CONSTRAINT_NAME asc, KCU.COLUMN_NAME asc",
+        "SysTableTableName": "[VARBINARY(\"data_type_table\"), VARBINARY(\"data_type_table\")]",
+        "SysTableTableSchema": "[VARBINARY(\"test\")]",
+        "Table": "INFORMATION_SCHEMA.KEY_COLUMN_USAGE, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS"
+      },
+      {
+        "OperatorType": "Route",
+        "Variant": "SelectDBA",
+        "Keyspace": {
+          "Name": "main",
+          "Sharded": false
+        },
+        "FieldQuery": "select S.UPDATE_RULE from INFORMATION_SCHEMA.K as S where 1 != 1",
+        "Query": "select S.UPDATE_RULE from INFORMATION_SCHEMA.K as S where S.TABLE_SCHEMA = :__vtschemaname and S.TABLE_NAME = :__vttablename",
+        "SysTableTableName": "[VARBINARY(\"sc\")]",
+        "SysTableTableSchema": "[VARBINARY(\"test\")]",
+        "Table": "INFORMATION_SCHEMA.K"
+      }
+    ]
+  }
+}
+
+#information_schema.routines
+"SELECT routine_name AS name, routine_definition AS definition FROM information_schema.routines WHERE ROUTINE_SCHEMA = ? AND ROUTINE_TYPE = 'PROCEDURE'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT routine_name AS name, routine_definition AS definition FROM information_schema.routines WHERE ROUTINE_SCHEMA = ? AND ROUTINE_TYPE = 'PROCEDURE'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select routine_name as `name`, routine_definition as definition from information_schema.routines where 1 != 1",
+    "Query": "select routine_name as `name`, routine_definition as definition from information_schema.routines where ROUTINE_SCHEMA = :__vtschemaname and ROUTINE_TYPE = 'PROCEDURE'",
+    "SysTableTableSchema": "[:v1]",
+    "Table": "information_schema.routines"
+  }
+}
+
+#information_schema table sizes
+"SELECT SUM(data_length + index_length) as size FROM information_schema.TABLES WHERE table_schema = ?"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT SUM(data_length + index_length) as size FROM information_schema.TABLES WHERE table_schema = ?",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select SUM(data_length + index_length) as size from information_schema.`TABLES` where 1 != 1",
+    "Query": "select SUM(data_length + index_length) as size from information_schema.`TABLES` where table_schema = :__vtschemaname",
+    "SysTableTableSchema": "[:v1]",
+    "Table": "information_schema.`TABLES`"
+  }
+}
+
+#information_schema referential contraints
+"SELECT kcu.constraint_name constraint_name, kcu.column_name column_name, kcu.referenced_table_name referenced_table_name, kcu.referenced_column_name referenced_column_name, kcu.ordinal_position ordinal_position, kcu.table_name table_name, rc.delete_rule delete_rule, rc.update_rule update_rule FROM information_schema.key_column_usage AS kcu INNER JOIN information_schema.referential_constraints AS rc ON kcu.constraint_name = rc.constraint_name WHERE kcu.table_schema = ? AND rc.constraint_schema = ? AND kcu.referenced_column_name IS NOT NULL ORDER BY ordinal_position"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT kcu.constraint_name constraint_name, kcu.column_name column_name, kcu.referenced_table_name referenced_table_name, kcu.referenced_column_name referenced_column_name, kcu.ordinal_position ordinal_position, kcu.table_name table_name, rc.delete_rule delete_rule, rc.update_rule update_rule FROM information_schema.key_column_usage AS kcu INNER JOIN information_schema.referential_constraints AS rc ON kcu.constraint_name = rc.constraint_name WHERE kcu.table_schema = ? AND rc.constraint_schema = ? AND kcu.referenced_column_name IS NOT NULL ORDER BY ordinal_position",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select kcu.constraint_name as constraint_name, kcu.column_name as column_name, kcu.referenced_table_name as referenced_table_name, kcu.referenced_column_name as referenced_column_name, kcu.ordinal_position as ordinal_position, kcu.table_name as table_name, rc.delete_rule as delete_rule, rc.update_rule as update_rule from information_schema.key_column_usage as kcu join information_schema.referential_constraints as rc on kcu.constraint_name = rc.constraint_name where 1 != 1",
+    "Query": "select kcu.constraint_name as constraint_name, kcu.column_name as column_name, kcu.referenced_table_name as referenced_table_name, kcu.referenced_column_name as referenced_column_name, kcu.ordinal_position as ordinal_position, kcu.table_name as table_name, rc.delete_rule as delete_rule, rc.update_rule as update_rule from information_schema.key_column_usage as kcu join information_schema.referential_constraints as rc on kcu.constraint_name = rc.constraint_name where kcu.table_schema = :__vtschemaname and rc.constraint_schema = :__vtschemaname and kcu.referenced_column_name is not null order by ordinal_position asc",
+    "SysTableTableSchema": "[:v1, :v2]",
+    "Table": "information_schema.key_column_usage, information_schema.referential_constraints"
+  }
+}
diff --git a/go/vt/vtgate/planbuilder/testdata/set_sysvar_disabled_cases.txt b/go/vt/vtgate/planbuilder/testdata/set_sysvar_disabled_cases.txt
index 46cbdfd4e0..f213ba1a96 100644
--- a/go/vt/vtgate/planbuilder/testdata/set_sysvar_disabled_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/set_sysvar_disabled_cases.txt
@@ -34,5 +34,3 @@
     ]
   }
 }
-Gen4 plan same as above
-
diff --git a/go/vt/vtgate/planbuilder/testdata/sysschema_default.txt b/go/vt/vtgate/planbuilder/testdata/sysschema_default.txt
index 0fc59247f6..f6e58dd2c1 100644
--- a/go/vt/vtgate/planbuilder/testdata/sysschema_default.txt
+++ b/go/vt/vtgate/planbuilder/testdata/sysschema_default.txt
@@ -15,39 +15,3 @@
     "Table": "dual"
   }
 }
-Gen4 plan same as above
-
-# unqualified table name
-"select t.table_schema,t.table_name,c.column_name,c.column_type from tables t join columns c on c.table_schema = t.table_schema and c.table_name = t.table_name where t.table_schema = 'user' and c.table_schema = 'user' order by t.table_schema,t.table_name,c.column_name"
-{
-  "QueryType": "SELECT",
-  "Original": "select t.table_schema,t.table_name,c.column_name,c.column_type from tables t join columns c on c.table_schema = t.table_schema and c.table_name = t.table_name where t.table_schema = 'user' and c.table_schema = 'user' order by t.table_schema,t.table_name,c.column_name",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select t.table_schema, t.table_name, c.column_name, c.column_type from information_schema.`tables` as t join information_schema.`columns` as c on c.table_schema = t.table_schema and c.table_name = t.table_name where 1 != 1",
-    "Query": "select t.table_schema, t.table_name, c.column_name, c.column_type from information_schema.`tables` as t join information_schema.`columns` as c on c.table_schema = t.table_schema and c.table_name = t.table_name where t.table_schema = :__vtschemaname and c.table_schema = :__vtschemaname order by t.table_schema asc, t.table_name asc, c.column_name asc",
-    "SysTableTableSchema": "[VARBINARY(\"user\"), VARBINARY(\"user\")]",
-    "Table": "information_schema.`tables`, information_schema.`columns`"
-  }
-}
-{
-  "QueryType": "SELECT",
-  "Original": "select t.table_schema,t.table_name,c.column_name,c.column_type from tables t join columns c on c.table_schema = t.table_schema and c.table_name = t.table_name where t.table_schema = 'user' and c.table_schema = 'user' order by t.table_schema,t.table_name,c.column_name",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select t.table_schema, t.table_name, c.column_name, c.column_type from information_schema.`tables` as t, information_schema.`columns` as c where 1 != 1",
-    "Query": "select t.table_schema, t.table_name, c.column_name, c.column_type from information_schema.`tables` as t, information_schema.`columns` as c where t.table_schema = :__vtschemaname and c.table_schema = :__vtschemaname and c.table_schema = t.table_schema and c.table_name = t.table_name order by t.table_schema asc, t.table_name asc, c.column_name asc",
-    "SysTableTableSchema": "[VARBINARY(\"user\"), VARBINARY(\"user\")]",
-    "Table": "information_schema.`columns`, information_schema.`tables`"
-  }
-}
diff --git a/go/vt/vtgate/planbuilder/testdata/systemtables_cases.txt b/go/vt/vtgate/planbuilder/testdata/systemtables_cases.txt
deleted file mode 100644
index ffbe3fe54d..0000000000
--- a/go/vt/vtgate/planbuilder/testdata/systemtables_cases.txt
+++ /dev/null
@@ -1,760 +0,0 @@
-# Single information_schema query
-"select col from information_schema.foo"
-{
-  "QueryType": "SELECT",
-  "Original": "select col from information_schema.foo",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select col from information_schema.foo where 1 != 1",
-    "Query": "select col from information_schema.foo",
-    "Table": "information_schema.foo"
-  }
-}
-Gen4 plan same as above
-
-# ',' join information_schema
-"select a.id,b.id from information_schema.a as a, information_schema.b as b"
-{
-  "QueryType": "SELECT",
-  "Original": "select a.id,b.id from information_schema.a as a, information_schema.b as b",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select a.id, b.id from information_schema.a as a, information_schema.b as b where 1 != 1",
-    "Query": "select a.id, b.id from information_schema.a as a, information_schema.b as b",
-    "Table": "information_schema.a, information_schema.b"
-  }
-}
-Gen4 plan same as above
-
-# information schema query that uses table_schema
-"select column_name from information_schema.columns where table_schema = (select schema())"
-{
-  "QueryType": "SELECT",
-  "Original": "select column_name from information_schema.columns where table_schema = (select schema())",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select column_name from information_schema.`columns` where 1 != 1",
-    "Query": "select column_name from information_schema.`columns` where table_schema = schema()",
-    "Table": "information_schema.`columns`"
-  }
-}
-Gen4 plan same as above
-
-# information schema join
-"select * from information_schema.a join information_schema.b"
-{
-  "QueryType": "SELECT",
-  "Original": "select * from information_schema.a join information_schema.b",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from information_schema.a join information_schema.b where 1 != 1",
-    "Query": "select * from information_schema.a join information_schema.b",
-    "Table": "information_schema.a, information_schema.b"
-  }
-}
-{
-  "QueryType": "SELECT",
-  "Original": "select * from information_schema.a join information_schema.b",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from information_schema.a, information_schema.b where 1 != 1",
-    "Query": "select * from information_schema.a, information_schema.b",
-    "Table": "information_schema.a, information_schema.b"
-  }
-}
-
-# access to unqualified column names in information_schema
-"select * from information_schema.a where b=10"
-{
-  "QueryType": "SELECT",
-  "Original": "select * from information_schema.a where b=10",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from information_schema.a where 1 != 1",
-    "Query": "select * from information_schema.a where b = 10",
-    "Table": "information_schema.a"
-  }
-}
-Gen4 plan same as above
-
-# access to qualified column names in information_schema
-"select * from information_schema.a where information_schema.a.b=10"
-{
-  "QueryType": "SELECT",
-  "Original": "select * from information_schema.a where information_schema.a.b=10",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from information_schema.a where 1 != 1",
-    "Query": "select * from information_schema.a where information_schema.a.b = 10",
-    "Table": "information_schema.a"
-  }
-}
-Gen4 plan same as above
-
-# union of information_schema
-"select * from information_schema.a union select * from information_schema.b"
-{
-  "QueryType": "SELECT",
-  "Original": "select * from information_schema.a union select * from information_schema.b",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from information_schema.a where 1 != 1 union select * from information_schema.b where 1 != 1",
-    "Query": "select * from information_schema.a union select * from information_schema.b",
-    "Table": "information_schema.a"
-  }
-}
-Gen4 plan same as above
-
-# Select from information schema query with two tables that route should be merged
-"SELECT DELETE_RULE, UPDATE_RULE FROM  INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS KCU INNER JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS RC ON KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME WHERE KCU.TABLE_SCHEMA = 'test' AND KCU.TABLE_NAME = 'data_type_table' AND KCU.COLUMN_NAME = 'id' AND KCU.REFERENCED_TABLE_SCHEMA = 'test' AND KCU.CONSTRAINT_NAME = 'data_type_table_id_fkey' ORDER BY KCU.CONSTRAINT_NAME, KCU.COLUMN_NAME"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT DELETE_RULE, UPDATE_RULE FROM  INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS KCU INNER JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS RC ON KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME WHERE KCU.TABLE_SCHEMA = 'test' AND KCU.TABLE_NAME = 'data_type_table' AND KCU.COLUMN_NAME = 'id' AND KCU.REFERENCED_TABLE_SCHEMA = 'test' AND KCU.CONSTRAINT_NAME = 'data_type_table_id_fkey' ORDER BY KCU.CONSTRAINT_NAME, KCU.COLUMN_NAME",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select DELETE_RULE, UPDATE_RULE from INFORMATION_SCHEMA.KEY_COLUMN_USAGE as KCU join INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS as RC on KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME where 1 != 1",
-    "Query": "select DELETE_RULE, UPDATE_RULE from INFORMATION_SCHEMA.KEY_COLUMN_USAGE as KCU join INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS as RC on KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME where KCU.TABLE_SCHEMA = :__vtschemaname and KCU.TABLE_NAME = :KCU_TABLE_NAME and KCU.COLUMN_NAME = 'id' and KCU.REFERENCED_TABLE_SCHEMA = 'test' and KCU.CONSTRAINT_NAME = 'data_type_table_id_fkey' order by KCU.CONSTRAINT_NAME asc, KCU.COLUMN_NAME asc",
-    "SysTableTableName": "[KCU_TABLE_NAME:VARBINARY(\"data_type_table\")]",
-    "SysTableTableSchema": "[VARBINARY(\"test\")]",
-    "Table": "INFORMATION_SCHEMA.KEY_COLUMN_USAGE, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS"
-  }
-}
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT DELETE_RULE, UPDATE_RULE FROM  INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS KCU INNER JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS RC ON KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME WHERE KCU.TABLE_SCHEMA = 'test' AND KCU.TABLE_NAME = 'data_type_table' AND KCU.COLUMN_NAME = 'id' AND KCU.REFERENCED_TABLE_SCHEMA = 'test' AND KCU.CONSTRAINT_NAME = 'data_type_table_id_fkey' ORDER BY KCU.CONSTRAINT_NAME, KCU.COLUMN_NAME",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select DELETE_RULE, UPDATE_RULE from INFORMATION_SCHEMA.KEY_COLUMN_USAGE as KCU, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS as RC where 1 != 1",
-    "Query": "select DELETE_RULE, UPDATE_RULE from INFORMATION_SCHEMA.KEY_COLUMN_USAGE as KCU, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS as RC where KCU.TABLE_SCHEMA = :__vtschemaname and KCU.TABLE_NAME = :KCU_TABLE_NAME and KCU.COLUMN_NAME = 'id' and KCU.REFERENCED_TABLE_SCHEMA = 'test' and KCU.CONSTRAINT_NAME = 'data_type_table_id_fkey' and KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME order by KCU.CONSTRAINT_NAME asc, KCU.COLUMN_NAME asc",
-    "SysTableTableName": "[KCU_TABLE_NAME:VARBINARY(\"data_type_table\")]",
-    "SysTableTableSchema": "[VARBINARY(\"test\")]",
-    "Table": "INFORMATION_SCHEMA.KEY_COLUMN_USAGE, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS"
-  }
-}
-
-# Select from information schema query with three tables such that route for 2 should be merged but not for the last.
-"SELECT KCU.DELETE_RULE, S.UPDATE_RULE FROM  INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS KCU INNER JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS RC ON KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME, INFORMATION_SCHEMA.K AS S WHERE KCU.TABLE_SCHEMA = 'test' AND KCU.TABLE_NAME = 'data_type_table' AND KCU.TABLE_NAME = 'data_type_table' AND S.TABLE_SCHEMA = 'test' AND S.TABLE_NAME = 'sc' ORDER BY KCU.CONSTRAINT_NAME, KCU.COLUMN_NAME"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT KCU.DELETE_RULE, S.UPDATE_RULE FROM  INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS KCU INNER JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS RC ON KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME, INFORMATION_SCHEMA.K AS S WHERE KCU.TABLE_SCHEMA = 'test' AND KCU.TABLE_NAME = 'data_type_table' AND KCU.TABLE_NAME = 'data_type_table' AND S.TABLE_SCHEMA = 'test' AND S.TABLE_NAME = 'sc' ORDER BY KCU.CONSTRAINT_NAME, KCU.COLUMN_NAME",
-  "Instructions": {
-    "OperatorType": "Join",
-    "Variant": "Join",
-    "JoinColumnIndexes": "-1,1",
-    "TableName": "INFORMATION_SCHEMA.KEY_COLUMN_USAGE, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS_INFORMATION_SCHEMA.K",
-    "Inputs": [
-      {
-        "OperatorType": "Route",
-        "Variant": "SelectDBA",
-        "Keyspace": {
-          "Name": "main",
-          "Sharded": false
-        },
-        "FieldQuery": "select KCU.DELETE_RULE from INFORMATION_SCHEMA.KEY_COLUMN_USAGE as KCU join INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS as RC on KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME where 1 != 1",
-        "Query": "select KCU.DELETE_RULE from INFORMATION_SCHEMA.KEY_COLUMN_USAGE as KCU join INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS as RC on KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME where KCU.TABLE_SCHEMA = :__vtschemaname and KCU.TABLE_NAME = :KCU_TABLE_NAME and KCU.TABLE_NAME = :KCU_TABLE_NAME order by KCU.CONSTRAINT_NAME asc, KCU.COLUMN_NAME asc",
-        "SysTableTableName": "[KCU_TABLE_NAME:VARBINARY(\"data_type_table\")]",
-        "SysTableTableSchema": "[VARBINARY(\"test\")]",
-        "Table": "INFORMATION_SCHEMA.KEY_COLUMN_USAGE, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS"
-      },
-      {
-        "OperatorType": "Route",
-        "Variant": "SelectDBA",
-        "Keyspace": {
-          "Name": "main",
-          "Sharded": false
-        },
-        "FieldQuery": "select S.UPDATE_RULE from INFORMATION_SCHEMA.K as S where 1 != 1",
-        "Query": "select S.UPDATE_RULE from INFORMATION_SCHEMA.K as S where S.TABLE_SCHEMA = :__vtschemaname and S.TABLE_NAME = :S_TABLE_NAME",
-        "SysTableTableName": "[S_TABLE_NAME:VARBINARY(\"sc\")]",
-        "SysTableTableSchema": "[VARBINARY(\"test\")]",
-        "Table": "INFORMATION_SCHEMA.K"
-      }
-    ]
-  }
-}
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT KCU.DELETE_RULE, S.UPDATE_RULE FROM  INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS KCU INNER JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS RC ON KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME, INFORMATION_SCHEMA.K AS S WHERE KCU.TABLE_SCHEMA = 'test' AND KCU.TABLE_NAME = 'data_type_table' AND KCU.TABLE_NAME = 'data_type_table' AND S.TABLE_SCHEMA = 'test' AND S.TABLE_NAME = 'sc' ORDER BY KCU.CONSTRAINT_NAME, KCU.COLUMN_NAME",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select KCU.DELETE_RULE, S.UPDATE_RULE from INFORMATION_SCHEMA.KEY_COLUMN_USAGE as KCU, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS as RC, INFORMATION_SCHEMA.K as S where 1 != 1",
-    "Query": "select KCU.DELETE_RULE, S.UPDATE_RULE from INFORMATION_SCHEMA.KEY_COLUMN_USAGE as KCU, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS as RC, INFORMATION_SCHEMA.K as S where S.TABLE_SCHEMA = :__vtschemaname and S.TABLE_NAME = :S_TABLE_NAME and KCU.TABLE_SCHEMA = :__vtschemaname and KCU.TABLE_NAME = :KCU_TABLE_NAME and KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME order by KCU.CONSTRAINT_NAME asc, KCU.COLUMN_NAME asc",
-    "SysTableTableName": "[KCU_TABLE_NAME:VARBINARY(\"data_type_table\"), S_TABLE_NAME:VARBINARY(\"sc\")]",
-    "SysTableTableSchema": "[VARBINARY(\"test\"), VARBINARY(\"test\")]",
-    "Table": "INFORMATION_SCHEMA.K, INFORMATION_SCHEMA.KEY_COLUMN_USAGE, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS"
-  }
-}
-
-#information_schema.routines
-"SELECT routine_name AS name, routine_definition AS definition FROM information_schema.routines WHERE ROUTINE_SCHEMA = ? AND ROUTINE_TYPE = 'PROCEDURE'"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT routine_name AS name, routine_definition AS definition FROM information_schema.routines WHERE ROUTINE_SCHEMA = ? AND ROUTINE_TYPE = 'PROCEDURE'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select routine_name as `name`, routine_definition as definition from information_schema.routines where 1 != 1",
-    "Query": "select routine_name as `name`, routine_definition as definition from information_schema.routines where ROUTINE_SCHEMA = :__vtschemaname and ROUTINE_TYPE = 'PROCEDURE'",
-    "SysTableTableSchema": "[:v1]",
-    "Table": "information_schema.routines"
-  }
-}
-Gen4 plan same as above
-
-#information_schema table sizes
-"SELECT SUM(data_length + index_length) as size FROM information_schema.TABLES WHERE table_schema = ?"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT SUM(data_length + index_length) as size FROM information_schema.TABLES WHERE table_schema = ?",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select SUM(data_length + index_length) as size from information_schema.`TABLES` where 1 != 1",
-    "Query": "select SUM(data_length + index_length) as size from information_schema.`TABLES` where table_schema = :__vtschemaname",
-    "SysTableTableSchema": "[:v1]",
-    "Table": "information_schema.`TABLES`"
-  }
-}
-Gen4 plan same as above
-
-#information_schema referential contraints
-"SELECT kcu.constraint_name constraint_name, kcu.column_name column_name, kcu.referenced_table_name referenced_table_name, kcu.referenced_column_name referenced_column_name, kcu.ordinal_position ordinal_position, kcu.table_name table_name, rc.delete_rule delete_rule, rc.update_rule update_rule FROM information_schema.key_column_usage AS kcu INNER JOIN information_schema.referential_constraints AS rc ON kcu.constraint_name = rc.constraint_name WHERE kcu.table_schema = ? AND rc.constraint_schema = ? AND kcu.referenced_column_name IS NOT NULL ORDER BY ordinal_position"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT kcu.constraint_name constraint_name, kcu.column_name column_name, kcu.referenced_table_name referenced_table_name, kcu.referenced_column_name referenced_column_name, kcu.ordinal_position ordinal_position, kcu.table_name table_name, rc.delete_rule delete_rule, rc.update_rule update_rule FROM information_schema.key_column_usage AS kcu INNER JOIN information_schema.referential_constraints AS rc ON kcu.constraint_name = rc.constraint_name WHERE kcu.table_schema = ? AND rc.constraint_schema = ? AND kcu.referenced_column_name IS NOT NULL ORDER BY ordinal_position",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select kcu.constraint_name as constraint_name, kcu.column_name as column_name, kcu.referenced_table_name as referenced_table_name, kcu.referenced_column_name as referenced_column_name, kcu.ordinal_position as ordinal_position, kcu.table_name as table_name, rc.delete_rule as delete_rule, rc.update_rule as update_rule from information_schema.key_column_usage as kcu join information_schema.referential_constraints as rc on kcu.constraint_name = rc.constraint_name where 1 != 1",
-    "Query": "select kcu.constraint_name as constraint_name, kcu.column_name as column_name, kcu.referenced_table_name as referenced_table_name, kcu.referenced_column_name as referenced_column_name, kcu.ordinal_position as ordinal_position, kcu.table_name as table_name, rc.delete_rule as delete_rule, rc.update_rule as update_rule from information_schema.key_column_usage as kcu join information_schema.referential_constraints as rc on kcu.constraint_name = rc.constraint_name where kcu.table_schema = :__vtschemaname and rc.constraint_schema = :__vtschemaname and kcu.referenced_column_name is not null order by ordinal_position asc",
-    "SysTableTableSchema": "[:v1, :v2]",
-    "Table": "information_schema.key_column_usage, information_schema.referential_constraints"
-  }
-}
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT kcu.constraint_name constraint_name, kcu.column_name column_name, kcu.referenced_table_name referenced_table_name, kcu.referenced_column_name referenced_column_name, kcu.ordinal_position ordinal_position, kcu.table_name table_name, rc.delete_rule delete_rule, rc.update_rule update_rule FROM information_schema.key_column_usage AS kcu INNER JOIN information_schema.referential_constraints AS rc ON kcu.constraint_name = rc.constraint_name WHERE kcu.table_schema = ? AND rc.constraint_schema = ? AND kcu.referenced_column_name IS NOT NULL ORDER BY ordinal_position",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select kcu.constraint_name as constraint_name, kcu.column_name as column_name, kcu.referenced_table_name as referenced_table_name, kcu.referenced_column_name as referenced_column_name, kcu.ordinal_position as ordinal_position, kcu.table_name as table_name, rc.delete_rule as delete_rule, rc.update_rule as update_rule from information_schema.key_column_usage as kcu, information_schema.referential_constraints as rc where 1 != 1",
-    "Query": "select kcu.constraint_name as constraint_name, kcu.column_name as column_name, kcu.referenced_table_name as referenced_table_name, kcu.referenced_column_name as referenced_column_name, kcu.ordinal_position as ordinal_position, kcu.table_name as table_name, rc.delete_rule as delete_rule, rc.update_rule as update_rule from information_schema.key_column_usage as kcu, information_schema.referential_constraints as rc where kcu.table_schema = :__vtschemaname and kcu.referenced_column_name is not null and rc.constraint_schema = :__vtschemaname and kcu.constraint_name = rc.constraint_name order by ordinal_position asc",
-    "SysTableTableSchema": "[:v1, :v2]",
-    "Table": "information_schema.key_column_usage, information_schema.referential_constraints"
-  }
-}
-
-# rails query
-"select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as name, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk using (constraint_schema, constraint_name) where fk.referenced_column_name is not null and fk.table_schema = database() and fk.table_name = ':vtg1' and rc.constraint_schema = database() and rc.table_name = ':vtg1'"
-{
-  "QueryType": "SELECT",
-  "Original": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as name, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk using (constraint_schema, constraint_name) where fk.referenced_column_name is not null and fk.table_schema = database() and fk.table_name = ':vtg1' and rc.constraint_schema = database() and rc.table_name = ':vtg1'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk on rc.constraint_schema = fk.constraint_schema and rc.constraint_name = fk.constraint_name where 1 != 1",
-    "Query": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk on rc.constraint_schema = fk.constraint_schema and rc.constraint_name = fk.constraint_name where fk.referenced_column_name is not null and fk.table_schema = database() and fk.table_name = :fk_table_name and rc.constraint_schema = database() and rc.table_name = :rc_table_name",
-    "SysTableTableName": "[fk_table_name:VARBINARY(\":vtg1\"), rc_table_name:VARBINARY(\":vtg1\")]",
-    "Table": "information_schema.referential_constraints, information_schema.key_column_usage"
-  }
-}
-{
-  "QueryType": "SELECT",
-  "Original": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as name, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk using (constraint_schema, constraint_name) where fk.referenced_column_name is not null and fk.table_schema = database() and fk.table_name = ':vtg1' and rc.constraint_schema = database() and rc.table_name = ':vtg1'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc, information_schema.key_column_usage as fk where 1 != 1",
-    "Query": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc, information_schema.key_column_usage as fk where rc.constraint_schema = database() and rc.table_name = :rc_table_name and fk.referenced_column_name is not null and fk.table_schema = database() and fk.table_name = :fk_table_name and rc.constraint_schema = fk.constraint_schema and rc.constraint_name = fk.constraint_name",
-    "SysTableTableName": "[fk_table_name:VARBINARY(\":vtg1\"), rc_table_name:VARBINARY(\":vtg1\")]",
-    "Table": "information_schema.key_column_usage, information_schema.referential_constraints"
-  }
-}
-
-#rails_query 2
-"SELECT * FROM information_schema.schemata WHERE schema_name = 'user'"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT * FROM information_schema.schemata WHERE schema_name = 'user'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from information_schema.schemata where 1 != 1",
-    "Query": "select * from information_schema.schemata where schema_name = :__vtschemaname",
-    "SysTableTableSchema": "[VARBINARY(\"user\")]",
-    "Table": "information_schema.schemata"
-  }
-}
-Gen4 plan same as above
-
-#rails_query 3
-"SELECT table_comment FROM information_schema.tables WHERE table_schema = 'schema_name' AND table_name = 'table_name'"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT table_comment FROM information_schema.tables WHERE table_schema = 'schema_name' AND table_name = 'table_name'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select table_comment from information_schema.`tables` where 1 != 1",
-    "Query": "select table_comment from information_schema.`tables` where table_schema = :__vtschemaname and table_name = :table_name",
-    "SysTableTableName": "[table_name:VARBINARY(\"table_name\")]",
-    "SysTableTableSchema": "[VARBINARY(\"schema_name\")]",
-    "Table": "information_schema.`tables`"
-  }
-}
-Gen4 plan same as above
-
-#rails_query 4
-"SELECT fk.referenced_table_name AS 'to_table', fk.referenced_column_name AS 'primary_key',fk.column_name AS 'column',fk.constraint_name AS 'name',rc.update_rule AS 'on_update',rc.delete_rule AS 'on_delete' FROM information_schema.referential_constraints rc JOIN information_schema.key_column_usage fk USING (constraint_schema, constraint_name) WHERE fk.referenced_column_name IS NOT NULL AND fk.table_schema = 'table_schema' AND fk.table_name = 'table_name' AND rc.constraint_schema = 'table_schema' AND rc.table_name = 'table_name'"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT fk.referenced_table_name AS 'to_table', fk.referenced_column_name AS 'primary_key',fk.column_name AS 'column',fk.constraint_name AS 'name',rc.update_rule AS 'on_update',rc.delete_rule AS 'on_delete' FROM information_schema.referential_constraints rc JOIN information_schema.key_column_usage fk USING (constraint_schema, constraint_name) WHERE fk.referenced_column_name IS NOT NULL AND fk.table_schema = 'table_schema' AND fk.table_name = 'table_name' AND rc.constraint_schema = 'table_schema' AND rc.table_name = 'table_name'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk on rc.constraint_schema = fk.constraint_schema and rc.constraint_name = fk.constraint_name where 1 != 1",
-    "Query": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk on rc.constraint_schema = fk.constraint_schema and rc.constraint_name = fk.constraint_name where fk.referenced_column_name is not null and fk.table_schema = :__vtschemaname and fk.table_name = :fk_table_name and rc.constraint_schema = :__vtschemaname and rc.table_name = :rc_table_name",
-    "SysTableTableName": "[fk_table_name:VARBINARY(\"table_name\"), rc_table_name:VARBINARY(\"table_name\")]",
-    "SysTableTableSchema": "[VARBINARY(\"table_schema\"), VARBINARY(\"table_schema\")]",
-    "Table": "information_schema.referential_constraints, information_schema.key_column_usage"
-  }
-}
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT fk.referenced_table_name AS 'to_table', fk.referenced_column_name AS 'primary_key',fk.column_name AS 'column',fk.constraint_name AS 'name',rc.update_rule AS 'on_update',rc.delete_rule AS 'on_delete' FROM information_schema.referential_constraints rc JOIN information_schema.key_column_usage fk USING (constraint_schema, constraint_name) WHERE fk.referenced_column_name IS NOT NULL AND fk.table_schema = 'table_schema' AND fk.table_name = 'table_name' AND rc.constraint_schema = 'table_schema' AND rc.table_name = 'table_name'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc, information_schema.key_column_usage as fk where 1 != 1",
-    "Query": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc, information_schema.key_column_usage as fk where rc.constraint_schema = :__vtschemaname and rc.table_name = :rc_table_name and fk.referenced_column_name is not null and fk.table_schema = :__vtschemaname and fk.table_name = :fk_table_name and rc.constraint_schema = fk.constraint_schema and rc.constraint_name = fk.constraint_name",
-    "SysTableTableName": "[fk_table_name:VARBINARY(\"table_name\"), rc_table_name:VARBINARY(\"table_name\")]",
-    "SysTableTableSchema": "[VARBINARY(\"table_schema\"), VARBINARY(\"table_schema\")]",
-    "Table": "information_schema.key_column_usage, information_schema.referential_constraints"
-  }
-}
-
-#rails_query 5
-"SELECT cc.constraint_name AS 'name', cc.check_clause AS 'expression' FROM information_schema.check_constraints cc JOIN information_schema.table_constraints tc USING (constraint_schema, constraint_name) WHERE tc.table_schema = 'table_schema' AND tc.table_name = 'table_name' AND cc.constraint_schema = 'constraint_schema'"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT cc.constraint_name AS 'name', cc.check_clause AS 'expression' FROM information_schema.check_constraints cc JOIN information_schema.table_constraints tc USING (constraint_schema, constraint_name) WHERE tc.table_schema = 'table_schema' AND tc.table_name = 'table_name' AND cc.constraint_schema = 'constraint_schema'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select cc.constraint_name as `name`, cc.check_clause as expression from information_schema.check_constraints as cc join information_schema.table_constraints as tc on cc.constraint_schema = tc.constraint_schema and cc.constraint_name = tc.constraint_name where 1 != 1",
-    "Query": "select cc.constraint_name as `name`, cc.check_clause as expression from information_schema.check_constraints as cc join information_schema.table_constraints as tc on cc.constraint_schema = tc.constraint_schema and cc.constraint_name = tc.constraint_name where tc.table_schema = :__vtschemaname and tc.table_name = :tc_table_name and cc.constraint_schema = :__vtschemaname",
-    "SysTableTableName": "[tc_table_name:VARBINARY(\"table_name\")]",
-    "SysTableTableSchema": "[VARBINARY(\"table_schema\"), VARBINARY(\"constraint_schema\")]",
-    "Table": "information_schema.check_constraints, information_schema.table_constraints"
-  }
-}
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT cc.constraint_name AS 'name', cc.check_clause AS 'expression' FROM information_schema.check_constraints cc JOIN information_schema.table_constraints tc USING (constraint_schema, constraint_name) WHERE tc.table_schema = 'table_schema' AND tc.table_name = 'table_name' AND cc.constraint_schema = 'constraint_schema'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select cc.constraint_name as `name`, cc.check_clause as expression from information_schema.check_constraints as cc, information_schema.table_constraints as tc where 1 != 1",
-    "Query": "select cc.constraint_name as `name`, cc.check_clause as expression from information_schema.check_constraints as cc, information_schema.table_constraints as tc where cc.constraint_schema = :__vtschemaname and tc.table_schema = :__vtschemaname and tc.table_name = :tc_table_name and cc.constraint_schema = tc.constraint_schema and cc.constraint_name = tc.constraint_name",
-    "SysTableTableName": "[tc_table_name:VARBINARY(\"table_name\")]",
-    "SysTableTableSchema": "[VARBINARY(\"constraint_schema\"), VARBINARY(\"table_schema\")]",
-    "Table": "information_schema.check_constraints, information_schema.table_constraints"
-  }
-}
-
-#rails_query 6
-"SELECT column_name FROM information_schema.statistics WHERE index_name = 'PRIMARY' AND table_schema = 'table_schema' AND table_name = 'table_name' ORDER BY seq_in_index"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT column_name FROM information_schema.statistics WHERE index_name = 'PRIMARY' AND table_schema = 'table_schema' AND table_name = 'table_name' ORDER BY seq_in_index",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select column_name from information_schema.statistics where 1 != 1",
-    "Query": "select column_name from information_schema.statistics where index_name = 'PRIMARY' and table_schema = :__vtschemaname and table_name = :table_name order by seq_in_index asc",
-    "SysTableTableName": "[table_name:VARBINARY(\"table_name\")]",
-    "SysTableTableSchema": "[VARBINARY(\"table_schema\")]",
-    "Table": "information_schema.statistics"
-  }
-}
-Gen4 plan same as above
-
-#rails_query 7
-"SELECT generation_expression FROM information_schema.columns WHERE table_schema = 'table_schema' AND table_name = 'table_name' AND column_name = 'column_name'"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT generation_expression FROM information_schema.columns WHERE table_schema = 'table_schema' AND table_name = 'table_name' AND column_name = 'column_name'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select generation_expression from information_schema.`columns` where 1 != 1",
-    "Query": "select generation_expression from information_schema.`columns` where table_schema = :__vtschemaname and table_name = :table_name and column_name = 'column_name'",
-    "SysTableTableName": "[table_name:VARBINARY(\"table_name\")]",
-    "SysTableTableSchema": "[VARBINARY(\"table_schema\")]",
-    "Table": "information_schema.`columns`"
-  }
-}
-Gen4 plan same as above
-
-#rails_query 8
-"SELECT id FROM information_schema.processlist WHERE info LIKE '% FOR UPDATE'"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT id FROM information_schema.processlist WHERE info LIKE '% FOR UPDATE'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select id from information_schema.`processlist` where 1 != 1",
-    "Query": "select id from information_schema.`processlist` where info like '% FOR UPDATE'",
-    "Table": "information_schema.`processlist`"
-  }
-}
-Gen4 plan same as above
-
-#rails_query 9
-"SELECT table_name FROM (SELECT * FROM information_schema.tables WHERE table_schema = 'table_schema') _subquery"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT table_name FROM (SELECT * FROM information_schema.tables WHERE table_schema = 'table_schema') _subquery",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select table_name from (select * from information_schema.`tables` where 1 != 1) as _subquery where 1 != 1",
-    "Query": "select table_name from (select * from information_schema.`tables` where table_schema = :__vtschemaname) as _subquery",
-    "SysTableTableSchema": "[VARBINARY(\"table_schema\")]",
-    "Table": "information_schema.`tables`"
-  }
-}
-
-#rails_query 10
-"SELECT table_name FROM (SELECT * FROM information_schema.tables WHERE table_schema = 'table_schema') _subquery WHERE _subquery.table_type = 'table_type' AND _subquery.table_name = 'table_name'"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT table_name FROM (SELECT * FROM information_schema.tables WHERE table_schema = 'table_schema') _subquery WHERE _subquery.table_type = 'table_type' AND _subquery.table_name = 'table_name'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select table_name from (select * from information_schema.`tables` where 1 != 1) as _subquery where 1 != 1",
-    "Query": "select table_name from (select * from information_schema.`tables` where table_schema = :__vtschemaname) as _subquery where _subquery.table_type = 'table_type' and _subquery.table_name = :_subquery_table_name",
-    "SysTableTableName": "[_subquery_table_name:VARBINARY(\"table_name\")]",
-    "SysTableTableSchema": "[VARBINARY(\"table_schema\")]",
-    "Table": "information_schema.`tables`"
-  }
-}
-
-# two predicates specifying the database for the same table work if the database is the same
-"SELECT cc.constraint_name AS 'name' FROM information_schema.check_constraints cc  WHERE cc.constraint_schema = 'a' AND cc.table_schema = 'a'"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT cc.constraint_name AS 'name' FROM information_schema.check_constraints cc  WHERE cc.constraint_schema = 'a' AND cc.table_schema = 'a'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select cc.constraint_name as `name` from information_schema.check_constraints as cc where 1 != 1",
-    "Query": "select cc.constraint_name as `name` from information_schema.check_constraints as cc where cc.constraint_schema = :__vtschemaname and cc.table_schema = :__vtschemaname",
-    "SysTableTableSchema": "[VARBINARY(\"a\"), VARBINARY(\"a\")]",
-    "Table": "information_schema.check_constraints"
-  }
-}
-Gen4 plan same as above
-
-# system schema in where clause of information_schema query
-"SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE table_schema = 'performance_schema' AND table_name = 'foo'"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE table_schema = 'performance_schema' AND table_name = 'foo'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select COUNT(*) from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
-    "Query": "select COUNT(*) from INFORMATION_SCHEMA.`TABLES` where table_schema = :__vtschemaname and table_name = :table_name",
-    "SysTableTableName": "[table_name:VARBINARY(\"foo\")]",
-    "SysTableTableSchema": "[VARBINARY(\"performance_schema\")]",
-    "Table": "INFORMATION_SCHEMA.`TABLES`"
-  }
-}
-Gen4 plan same as above
-
-# subquery of information_schema with itself
-"select * from information_schema.a where id in (select * from information_schema.b)"
-{
-  "QueryType": "SELECT",
-  "Original": "select * from information_schema.a where id in (select * from information_schema.b)",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from information_schema.a where 1 != 1",
-    "Query": "select * from information_schema.a where id in (select * from information_schema.b)",
-    "Table": "information_schema.a"
-  }
-}
-
-# query trying to query two different keyspaces at the same time
-"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'user' AND TABLE_SCHEMA = 'main'"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'user' AND TABLE_SCHEMA = 'main'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
-    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = :__vtschemaname and TABLE_SCHEMA = :__vtschemaname",
-    "SysTableTableSchema": "[VARBINARY(\"user\"), VARBINARY(\"main\")]",
-    "Table": "INFORMATION_SCHEMA.`TABLES`"
-  }
-}
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'user' AND TABLE_SCHEMA = 'main'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
-    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = :__vtschemaname",
-    "SysTableTableSchema": "[VARBINARY(\"user\"), VARBINARY(\"main\")]",
-    "Table": "INFORMATION_SCHEMA.`TABLES`"
-  }
-}
-
-# information_schema query using database() func
-"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = database()"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = database()",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
-    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = database()",
-    "Table": "INFORMATION_SCHEMA.`TABLES`"
-  }
-}
-Gen4 plan same as above
-
-# table_schema predicate the wrong way around
-"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE 'ks' = TABLE_SCHEMA"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE 'ks' = TABLE_SCHEMA",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
-    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = :__vtschemaname",
-    "SysTableTableSchema": "[VARBINARY(\"ks\")]",
-    "Table": "INFORMATION_SCHEMA.`TABLES`"
-  }
-}
-Gen4 plan same as above
-
-# table_name predicate against a routed table
-"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'ks' AND TABLE_NAME = 'route1'"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'ks' AND TABLE_NAME = 'route1'",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
-    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = :__vtschemaname and TABLE_NAME = :TABLE_NAME",
-    "SysTableTableName": "[TABLE_NAME:VARBINARY(\"route1\")]",
-    "SysTableTableSchema": "[VARBINARY(\"ks\")]",
-    "Table": "INFORMATION_SCHEMA.`TABLES`"
-  }
-}
-Gen4 plan same as above
-
-# information_schema query with additional predicates
-"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'ks' and other_column = 42"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'ks' and other_column = 42",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
-    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = :__vtschemaname and other_column = 42",
-    "SysTableTableSchema": "[VARBINARY(\"ks\")]",
-    "Table": "INFORMATION_SCHEMA.`TABLES`"
-  }
-}
-Gen4 plan same as above
-
-# able to isolate table_schema value even when hidden inside of ORs
-"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE (TABLE_SCHEMA = 'ks' and other_column = 42) OR (TABLE_SCHEMA = 'ks' and foobar = 'value')"
-{
-  "QueryType": "SELECT",
-  "Original": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE (TABLE_SCHEMA = 'ks' and other_column = 42) OR (TABLE_SCHEMA = 'ks' and foobar = 'value')",
-  "Instructions": {
-    "OperatorType": "Route",
-    "Variant": "SelectDBA",
-    "Keyspace": {
-      "Name": "main",
-      "Sharded": false
-    },
-    "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
-    "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = :__vtschemaname and (other_column = 42 or TABLE_SCHEMA = 'ks') and (other_column = 42 or foobar = 'value')",
-    "SysTableTableSchema": "[VARBINARY(\"ks\")]",
-    "Table": "INFORMATION_SCHEMA.`TABLES`"
-  }
-}
-Gen4 plan same as above
diff --git a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
index 5d32673c74..f8d3506995 100644
--- a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
@@ -421,6 +421,26 @@ Gen4 plan same as above
 "select * from user into outfile s3 'out_file_name'"
 "INTO is not supported on sharded keyspace"
 
+# unsupported two predicates specifying the database for the same table if they are different
+# will fail on run time but will pass on the planbuilder
+"SELECT cc.constraint_name AS 'name' FROM information_schema.check_constraints cc  WHERE cc.constraint_schema = 'constraint_schema' AND cc.table_schema = 'a'"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT cc.constraint_name AS 'name' FROM information_schema.check_constraints cc  WHERE cc.constraint_schema = 'constraint_schema' AND cc.table_schema = 'a'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "SelectDBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select cc.constraint_name as `name` from information_schema.check_constraints as cc where 1 != 1",
+    "Query": "select cc.constraint_name as `name` from information_schema.check_constraints as cc where cc.constraint_schema = :__vtschemaname and cc.table_schema = :__vtschemaname",
+    "SysTableTableSchema": "[VARBINARY(\"constraint_schema\"), VARBINARY(\"a\")]",
+    "Table": "information_schema.check_constraints"
+  }
+}
+
 # create view with Cannot auto-resolve for cross-shard joins
 "create view user.view_a as select col from user join user_extra"
 "symbol col not found"