Repository: vitessio/vitess
Pull request #9549 (2022-01-21 17:02:47)

Fix text handling in Vitess native evaluation
## Description

This PR started as an straightforward implementation of a couple functions for the `evalengine` (`COALESCE`, `LEAST` and `GREATEST`), but while implementing this I discovered that there was an underlying issue in the way Vitess was handling `Value` types from MySQL and that was preventing me from implementing these APIs cleanly: both when normalizing, and when interacting with bind variables, Vitess was coercing all byte-sequence data types to `VARBINARY` through the codebase, even those that MySQL tagged as representing "text" and originally sent as `VARCHAR`.

Hence, this PR overhauls this handling throughout the codebase so we can more accurately match MySQLs behavior. If you review the commits, you'll see the there are actually very few changes to the code: most of the fixes are trivial replacements in tests that were expecting to receive text values with a `VARBINARY` type and are now (correctly) receiving them as `VARCHAR`.

The only notable change is https://github.com/vitessio/vitess/pull/9549/commits/817fe5485c43d8535307a9bd3f1d3a8ee8d11181, in the `vstreamer` code. During our collation work this last year, we left some TODOs about wiring up collations throughout the codebase -- I've fixed one of them here because previously VStreamer filtering operators never handled "text" (they always compared using binary comparison because they saw as columns are being binary), but now textual columns are properly compared using collations and hence need to know what collation to use for comparison. cc @rohit-nayak-ps for review

The other major change that this PR introduces are planning changes for some query plans. These are well expected, most notably we can now plan a type of query with `COALESCE` which we couldn't handle before. @frouioui has reviewed and approved the new plans already.

**This PR introduces a minor breaking change to the `vtgate`s**, which is why I'd like to ship it before Tuesday's RC1 tagging: certain queries (most notably, selecting system variables) that previously returned from VTGate as `VARBINARY` blobs will now be `VARCHAR` instead. **This matches MySQL's behavior** (while previously, it didn't), and it shouldn't have an visible effect on the vast majority of MySQL drivers/connectors/clients because these perform coercion on the client to whatever type the user has requested. Furthermore, a lot of dynamic programming languages that would see a difference don't really have different types to differentiate between "bytes" and "string", and were already converting all `VARBINARY` rows to strings anyway, so they will see no difference here either.

## Related Issue(s)
<!-- List related issues and pull requests. If this PR fixes an issue, please add it using Fixes #????  -->


## Checklist
- [ ] Should this PR be backported?
- [x] Tests were added or are not required
- [x] Documentation was added or is not required

## Deployment Notes
<!-- Notes regarding deployment of the contained body of work. These should note any db migrations, etc. -->


Patch:
diff --git a/go/sqltypes/bind_variables.go b/go/sqltypes/bind_variables.go
index b6dc6f134b..4083f873ec 100644
--- a/go/sqltypes/bind_variables.go
+++ b/go/sqltypes/bind_variables.go
@@ -117,7 +117,7 @@ func DecimalBindVariable(v DecimalFloat) *querypb.BindVariable {
 
 // StringBindVariable converts a string to a bind var.
 func StringBindVariable(v string) *querypb.BindVariable {
-	return ValueBindVariable(NewVarBinary(v))
+	return ValueBindVariable(NewVarChar(v))
 }
 
 // BytesBindVariable converts a []byte to a bind var.
@@ -134,7 +134,7 @@ func ValueBindVariable(v Value) *querypb.BindVariable {
 func BuildBindVariable(v interface{}) (*querypb.BindVariable, error) {
 	switch v := v.(type) {
 	case string:
-		return BytesBindVariable([]byte(v)), nil
+		return StringBindVariable(v), nil
 	case []byte:
 		return BytesBindVariable(v), nil
 	case bool:
@@ -184,7 +184,7 @@ func BuildBindVariable(v interface{}) (*querypb.BindVariable, error) {
 		}
 		values := make([]querypb.Value, len(v))
 		for i, lv := range v {
-			values[i].Type = querypb.Type_VARBINARY
+			values[i].Type = querypb.Type_VARCHAR
 			values[i].Value = []byte(lv)
 			bv.Values[i] = &values[i]
 		}
diff --git a/go/vt/sqlparser/normalizer.go b/go/vt/sqlparser/normalizer.go
index f436803825..246ee438a3 100644
--- a/go/vt/sqlparser/normalizer.go
+++ b/go/vt/sqlparser/normalizer.go
@@ -118,7 +118,7 @@ func (nz *normalizer) convertLiteralDedup(node *Literal, cursor *Cursor) {
 
 	// Check if there's a bindvar for that value already.
 	var key string
-	if bval.Type == sqltypes.VarBinary {
+	if bval.Type == sqltypes.VarBinary || bval.Type == sqltypes.VarChar {
 		// Prefixing strings with "'" ensures that a string
 		// and number that have the same representation don't
 		// collide.
@@ -191,7 +191,7 @@ func (nz *normalizer) sqlToBindvar(node SQLNode) *querypb.BindVariable {
 		var err error
 		switch node.Type {
 		case StrVal:
-			v, err = sqltypes.NewValue(sqltypes.VarBinary, node.Bytes())
+			v, err = sqltypes.NewValue(sqltypes.VarChar, node.Bytes())
 		case IntVal:
 			v, err = sqltypes.NewValue(sqltypes.Int64, node.Bytes())
 		case FloatVal:
diff --git a/go/vt/vtgate/evalengine/arithmetic_expr.go b/go/vt/vtgate/evalengine/arithmetic_expr.go
index a02e8f9726..805f88a2fb 100644
--- a/go/vt/vtgate/evalengine/arithmetic_expr.go
+++ b/go/vt/vtgate/evalengine/arithmetic_expr.go
@@ -17,7 +17,6 @@ limitations under the License.
 package evalengine
 
 import (
-	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/sqltypes"
 	querypb "vitess.io/vitess/go/vt/proto/query"
 )
@@ -46,10 +45,6 @@ var _ ArithmeticOp = (*OpSubstraction)(nil)
 var _ ArithmeticOp = (*OpMultiplication)(nil)
 var _ ArithmeticOp = (*OpDivision)(nil)
 
-func (b *ArithmeticExpr) collation() collations.TypedCollation {
-	return collationNumeric
-}
-
 func (b *ArithmeticExpr) eval(env *ExpressionEnv, out *EvalResult) {
 	var left, right EvalResult
 	left.init(env, b.Left)
diff --git a/go/vt/vtgate/evalengine/cached_size.go b/go/vt/vtgate/evalengine/cached_size.go
index c7e5eab48c..52af6589c8 100644
--- a/go/vt/vtgate/evalengine/cached_size.go
+++ b/go/vt/vtgate/evalengine/cached_size.go
@@ -45,18 +45,6 @@ func (cached *ArithmeticExpr) CachedSize(alloc bool) int64 {
 	}
 	return size
 }
-func (cached *BinaryCoercedExpr) CachedSize(alloc bool) int64 {
-	if cached == nil {
-		return int64(0)
-	}
-	size := int64(0)
-	if alloc {
-		size += int64(64)
-	}
-	// field BinaryExpr vitess.io/vitess/go/vt/vtgate/evalengine.BinaryExpr
-	size += cached.BinaryExpr.CachedSize(false)
-	return size
-}
 func (cached *BinaryExpr) CachedSize(alloc bool) int64 {
 	if cached == nil {
 		return int64(0)
@@ -87,6 +75,34 @@ func (cached *BindVariable) CachedSize(alloc bool) int64 {
 	size += hack.RuntimeAllocSize(int64(len(cached.Key)))
 	return size
 }
+func (cached *CallExpression) CachedSize(alloc bool) int64 {
+	if cached == nil {
+		return int64(0)
+	}
+	size := int64(0)
+	if alloc {
+		size += int64(80)
+	}
+	// field Arguments vitess.io/vitess/go/vt/vtgate/evalengine.TupleExpr
+	{
+		size += hack.RuntimeAllocSize(int64(cap(cached.Arguments)) * int64(16))
+		for _, elem := range cached.Arguments {
+			if cc, ok := elem.(cachedObject); ok {
+				size += cc.CachedSize(true)
+			}
+		}
+	}
+	// field Aliases []vitess.io/vitess/go/vt/sqlparser.ColIdent
+	{
+		size += hack.RuntimeAllocSize(int64(cap(cached.Aliases)) * int64(40))
+		for _, elem := range cached.Aliases {
+			size += elem.CachedSize(false)
+		}
+	}
+	// field Method string
+	size += hack.RuntimeAllocSize(int64(len(cached.Method)))
+	return size
+}
 func (cached *CollateExpr) CachedSize(alloc bool) int64 {
 	if cached == nil {
 		return int64(0)
@@ -115,10 +131,10 @@ func (cached *ComparisonExpr) CachedSize(alloc bool) int64 {
 	}
 	size := int64(0)
 	if alloc {
-		size += int64(80)
+		size += int64(48)
 	}
-	// field BinaryCoercedExpr vitess.io/vitess/go/vt/vtgate/evalengine.BinaryCoercedExpr
-	size += cached.BinaryCoercedExpr.CachedSize(false)
+	// field BinaryExpr vitess.io/vitess/go/vt/vtgate/evalengine.BinaryExpr
+	size += cached.BinaryExpr.CachedSize(false)
 	// field Op vitess.io/vitess/go/vt/vtgate/evalengine.ComparisonOp
 	if cc, ok := cached.Op.(cachedObject); ok {
 		size += cc.CachedSize(true)
@@ -232,10 +248,10 @@ func (cached *LikeExpr) CachedSize(alloc bool) int64 {
 	}
 	size := int64(0)
 	if alloc {
-		size += int64(80)
+		size += int64(64)
 	}
-	// field BinaryCoercedExpr vitess.io/vitess/go/vt/vtgate/evalengine.BinaryCoercedExpr
-	size += cached.BinaryCoercedExpr.CachedSize(false)
+	// field BinaryExpr vitess.io/vitess/go/vt/vtgate/evalengine.BinaryExpr
+	size += cached.BinaryExpr.CachedSize(false)
 	// field Match vitess.io/vitess/go/mysql/collations.WildcardPattern
 	if cc, ok := cached.Match.(cachedObject); ok {
 		size += cc.CachedSize(true)
diff --git a/go/vt/vtgate/evalengine/collate.go b/go/vt/vtgate/evalengine/collate.go
index ce386945e6..e51ebdde07 100644
--- a/go/vt/vtgate/evalengine/collate.go
+++ b/go/vt/vtgate/evalengine/collate.go
@@ -34,6 +34,12 @@ var collationNumeric = collations.TypedCollation{
 	Repertoire:   collations.RepertoireASCII,
 }
 
+var collationBinary = collations.TypedCollation{
+	Collation:    collations.CollationBinaryID,
+	Coercibility: collations.CoerceExplicit,
+	Repertoire:   collations.RepertoireASCII,
+}
+
 func (c *CollateExpr) eval(env *ExpressionEnv, out *EvalResult) {
 	out.init(env, c.Inner)
 	if err := collations.Local().EnsureCollate(out.collation().Collation, c.TypedCollation.Collation); err != nil {
@@ -41,24 +47,3 @@ func (c *CollateExpr) eval(env *ExpressionEnv, out *EvalResult) {
 	}
 	out.replaceCollation(c.TypedCollation)
 }
-
-func (c *CollateExpr) collation() collations.TypedCollation {
-	return c.TypedCollation
-}
-
-func (t TupleExpr) collation() collations.TypedCollation {
-	// a Tuple does not have a collation, but an individual collation for every element of the tuple
-	return collations.TypedCollation{}
-}
-
-func (l *Literal) collation() collations.TypedCollation {
-	return l.Val.collation()
-}
-
-func (bv *BindVariable) collation() collations.TypedCollation {
-	return bv.coll
-}
-
-func (c *Column) collation() collations.TypedCollation {
-	return c.coll
-}
diff --git a/go/vt/vtgate/evalengine/comparisons.go b/go/vt/vtgate/evalengine/comparisons.go
index 22558fb036..477f7fbdda 100644
--- a/go/vt/vtgate/evalengine/comparisons.go
+++ b/go/vt/vtgate/evalengine/comparisons.go
@@ -17,8 +17,6 @@ limitations under the License.
 package evalengine
 
 import (
-	"fmt"
-
 	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/sqltypes"
 	querypb "vitess.io/vitess/go/vt/proto/query"
@@ -27,21 +25,16 @@ import (
 )
 
 type (
-	BinaryCoercedExpr struct {
-		BinaryExpr
-		CoerceLeft, CoerceRight collations.Coercion
-		MergedCollation         collations.TypedCollation
-	}
-
 	ComparisonExpr struct {
-		BinaryCoercedExpr
+		BinaryExpr
 		Op ComparisonOp
 	}
 
 	LikeExpr struct {
-		BinaryCoercedExpr
-		Negate bool
-		Match  collations.WildcardPattern
+		BinaryExpr
+		Negate         bool
+		Match          collations.WildcardPattern
+		MatchCollation collations.ID
 	}
 
 	InExpr struct {
@@ -106,29 +99,6 @@ func (compareNullSafeEQ) compare(left, right *EvalResult) (boolean, error) {
 	return makeboolean(cmp), err
 }
 
-func (c *BinaryCoercedExpr) collation() collations.TypedCollation {
-	// the collation of a binary operation is always integer, not the shared collation
-	// between the two subexpressions
-	return collationNumeric
-}
-
-func (c *BinaryCoercedExpr) coerce() error {
-	var err error
-
-	leftColl := c.Left.collation()
-	rightColl := c.Right.collation()
-
-	if leftColl.Valid() && rightColl.Valid() {
-		env := collations.Local()
-		c.MergedCollation, c.CoerceLeft, c.CoerceRight, err =
-			env.MergeCollations(leftColl, rightColl, collations.CoercionOptions{
-				ConvertToSuperset:   true,
-				ConvertWithCoercion: true,
-			})
-	}
-	return err
-}
-
 func evalResultsAreStrings(l, r *EvalResult) bool {
 	return l.textual() && r.textual()
 }
@@ -386,12 +356,8 @@ func (i *InExpr) typeof(env *ExpressionEnv) querypb.Type {
 	return -1
 }
 
-func (i *InExpr) collation() collations.TypedCollation {
-	return collationNumeric
-}
-
 func (l *LikeExpr) matchWildcard(left, right []byte, coll collations.ID) bool {
-	if l.Match != nil {
+	if l.Match != nil && l.MatchCollation == coll {
 		return l.Match.Match(left)
 	}
 	fullColl := collations.Local().LookupByID(coll)
@@ -404,6 +370,10 @@ func (l *LikeExpr) eval(env *ExpressionEnv, result *EvalResult) {
 	left.init(env, l.Left)
 	right.init(env, l.Right)
 
+	if err := mergeCollations(&left, &right); err != nil {
+		throwEvalError(err)
+	}
+
 	var matched bool
 	var leftcoll = left.collation()
 	var rightcoll = right.collation()
@@ -411,24 +381,15 @@ func (l *LikeExpr) eval(env *ExpressionEnv, result *EvalResult) {
 	switch {
 	case left.typeof() == querypb.Type_TUPLE || right.typeof() == querypb.Type_TUPLE:
 		panic("failed to typecheck tuples")
-
 	case left.null() || right.null():
 		result.setNull()
 		return
-
 	case left.textual() && right.textual():
-		if leftcoll.Collation != rightcoll.Collation {
-			panic(fmt.Sprintf("LikeOp: did not coerce, left=%d right=%d",
-				leftcoll.Collation, rightcoll.Collation))
-		}
 		matched = l.matchWildcard(left.bytes(), right.bytes(), rightcoll.Collation)
-
 	case right.textual():
 		matched = l.matchWildcard(left.value().Raw(), right.bytes(), rightcoll.Collation)
-
 	case left.textual():
 		matched = l.matchWildcard(left.bytes(), right.value().Raw(), leftcoll.Collation)
-
 	default:
 		matched = l.matchWildcard(left.value().Raw(), right.value().Raw(), collations.CollationBinaryID)
 	}
@@ -442,7 +403,3 @@ func (l *LikeExpr) typeof(env *ExpressionEnv) querypb.Type {
 	// return querypb.Type_UINT64, nil
 	return -1
 }
-
-func (l *LikeExpr) collation() collations.TypedCollation {
-	return collationNumeric
-}
diff --git a/go/vt/vtgate/evalengine/convert.go b/go/vt/vtgate/evalengine/convert.go
index 4e92764fc0..6bee80c6b3 100644
--- a/go/vt/vtgate/evalengine/convert.go
+++ b/go/vt/vtgate/evalengine/convert.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	"vitess.io/vitess/go/mysql/collations"
+	querypb "vitess.io/vitess/go/vt/proto/query"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vterrors"
@@ -60,32 +61,25 @@ func convertComparisonExpr2(op sqlparser.ComparisonExprOperator, left, right Exp
 		}, nil
 	}
 
-	coercedExpr := BinaryCoercedExpr{
-		BinaryExpr: binaryExpr,
-	}
-	if err := coercedExpr.coerce(); err != nil {
-		return nil, err
-	}
-
 	switch op {
 	case sqlparser.EqualOp:
-		return &ComparisonExpr{coercedExpr, compareEQ{}}, nil
+		return &ComparisonExpr{binaryExpr, compareEQ{}}, nil
 	case sqlparser.NotEqualOp:
-		return &ComparisonExpr{coercedExpr, compareNE{}}, nil
+		return &ComparisonExpr{binaryExpr, compareNE{}}, nil
 	case sqlparser.LessThanOp:
-		return &ComparisonExpr{coercedExpr, compareLT{}}, nil
+		return &ComparisonExpr{binaryExpr, compareLT{}}, nil
 	case sqlparser.LessEqualOp:
-		return &ComparisonExpr{coercedExpr, compareLE{}}, nil
+		return &ComparisonExpr{binaryExpr, compareLE{}}, nil
 	case sqlparser.GreaterThanOp:
-		return &ComparisonExpr{coercedExpr, compareGT{}}, nil
+		return &ComparisonExpr{binaryExpr, compareGT{}}, nil
 	case sqlparser.GreaterEqualOp:
-		return &ComparisonExpr{coercedExpr, compareGE{}}, nil
+		return &ComparisonExpr{binaryExpr, compareGE{}}, nil
 	case sqlparser.NullSafeEqualOp:
-		return &ComparisonExpr{coercedExpr, compareNullSafeEQ{}}, nil
+		return &ComparisonExpr{binaryExpr, compareNullSafeEQ{}}, nil
 	case sqlparser.LikeOp:
-		return &LikeExpr{BinaryCoercedExpr: coercedExpr}, nil
+		return &LikeExpr{BinaryExpr: binaryExpr}, nil
 	case sqlparser.NotLikeOp:
-		return &LikeExpr{BinaryCoercedExpr: coercedExpr, Negate: true}, nil
+		return &LikeExpr{BinaryExpr: binaryExpr, Negate: true}, nil
 	default:
 		return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, op.ToString())
 	}
@@ -217,8 +211,8 @@ func convertExpr(e sqlparser.Expr, lookup ConverterLookup) (Expr, error) {
 		case sqlparser.StrVal:
 			collation := getCollation(e, lookup)
 			return NewLiteralString(node.Bytes(), collation), nil
-		case sqlparser.HexNum:
-			return nil, convertNotSupported(e)
+		case sqlparser.HexVal:
+			return NewLiteralBinaryFromHex(node.Bytes())
 		}
 	case sqlparser.BoolVal:
 		if node {
@@ -300,25 +294,81 @@ func convertExpr(e sqlparser.Expr, lookup ConverterLookup) (Expr, error) {
 		if err != nil {
 			return nil, err
 		}
-		coll := collations.Local().DefaultCollationForCharset(node.CharacterSet[1:])
-		if coll == nil {
-			panic(fmt.Sprintf("unknown character set: %s", node.CharacterSet))
+		var collation collations.ID
+
+		// TODO: case insensitive
+		if node.CharacterSet == "_binary" {
+			collation = collations.CollationBinaryID
+		} else {
+			defaultCollation := collations.Local().DefaultCollationForCharset(node.CharacterSet[1:])
+			if defaultCollation == nil {
+				panic(fmt.Sprintf("unknown character set: %s", node.CharacterSet))
+			}
+			collation = defaultCollation.ID()
 		}
+
 		switch lit := expr.(type) {
 		case *Literal:
-			lit.Val.replaceCollationID(coll.ID())
+			switch collation {
+			case collations.CollationBinaryID:
+				lit.Val.type_ = querypb.Type_VARBINARY
+				lit.Val.collation_ = collationBinary
+			default:
+				lit.Val.type_ = querypb.Type_VARCHAR
+				lit.Val.replaceCollationID(collation)
+			}
 		case *BindVariable:
-			lit.coll.Collation = coll.ID()
+			switch collation {
+			case collations.CollationBinaryID:
+				lit.coerceType = querypb.Type_VARBINARY
+				lit.coll = collationBinary
+			default:
+				lit.coerceType = querypb.Type_VARCHAR
+				lit.coll.Collation = collation
+			}
 		default:
 			panic("character set introducers are only supported for literals and arguments")
 		}
 		return expr, nil
 	case *sqlparser.IsExpr:
 		return convertIsExpr(node.Left, node.Right, lookup)
+	case *sqlparser.FuncExpr:
+		method := node.Name.Lowered()
+		call, ok := builtinFunctions[method]
+		if !ok {
+			break
+		}
+
+		var args TupleExpr
+		var aliases []sqlparser.ColIdent
+		for _, expr := range node.Exprs {
+			aliased, ok := expr.(*sqlparser.AliasedExpr)
+			if !ok {
+				return nil, convertNotSupported(e)
+			}
+			convertedExpr, err := convertExpr(aliased.Expr, lookup)
+			if err != nil {
+				return nil, err
+			}
+			args = append(args, convertedExpr)
+			aliases = append(aliases, aliased.As)
+		}
+		return &CallExpression{
+			Arguments: args,
+			Aliases:   aliases,
+			Method:    method,
+			Call:      call,
+		}, nil
 	}
 	return nil, convertNotSupported(e)
 }
 
+var builtinFunctions = map[string]func([]EvalResult, *EvalResult){
+	"coalesce": builtinFuncCoalesce,
+	"greatest": builtinFuncGreatest,
+	"least":    builtinFuncLeast,
+}
+
 func convertNotSupported(e sqlparser.Expr) error {
 	return vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "%s: %s", ErrConvertExprNotSupported, sqlparser.String(e))
 }
diff --git a/go/vt/vtgate/evalengine/eval_result.go b/go/vt/vtgate/evalengine/eval_result.go
index c638b86680..f806c7cbe5 100644
--- a/go/vt/vtgate/evalengine/eval_result.go
+++ b/go/vt/vtgate/evalengine/eval_result.go
@@ -17,6 +17,7 @@ limitations under the License.
 package evalengine
 
 import (
+	"bytes"
 	"fmt"
 	"math"
 	"strconv"
@@ -46,23 +47,23 @@ type (
 		// Must not be accessed directly: call EvalResult.typeof() instead.
 		// For most expression types, this is known ahead of time and calling typeof() does not require
 		// an evaluation, so the type of an expression can be known without evaluating it.
-		type_ querypb.Type
+		type_ querypb.Type //nolint
 		// collation_ is the collation of this result. It may be uninitialized.
 		// Must not be accessed directly: call EvalResult.collation() instead.
-		collation_ collations.TypedCollation
+		collation_ collations.TypedCollation //nolint
 		// numeric_ is the numeric value of this result. It may be uninitialized.
 		// Must not be accessed directly: call one of the numeric getters for EvalResult instead.
-		numeric_ uint64
+		numeric_ uint64 //nolint
 		// bytes_ is the raw byte value this result. It may be uninitialized.
 		// Must not be accessed directly: call EvalResult.bytes() instead.
-		bytes_ []byte
+		bytes_ []byte //nolint
 		// tuple_ is the list of all results contained in this result, if the result is a tuple.
 		// It may be uninitialized.
 		// Must not be accessed directly: call EvalResult.tuple() instead.
-		tuple_ *[]EvalResult
+		tuple_ *[]EvalResult //nolint
 		// decimal_ is the numeric decimal for this result. It may be uninitialized.
 		// Must not be accessed directly: call EvalResult.decimal() instead.
-		decimal_ *decimalResult
+		decimal_ *decimalResult //nolint
 	}
 
 	decimalResult struct {
@@ -147,7 +148,7 @@ func (er *EvalResult) string() string {
 }
 
 func (er *EvalResult) value() sqltypes.Value {
-	return er.toSQLValue(er.typeof())
+	return sqltypes.MakeTrusted(er.typeof(), er.toRawBytes())
 }
 
 func (er *EvalResult) null() bool {
@@ -183,6 +184,12 @@ func (er *EvalResult) setRaw(typ querypb.Type, raw []byte, coll collations.Typed
 	er.collation_ = coll
 }
 
+func (er *EvalResult) setRawNumeric(typ querypb.Type, u uint64) {
+	er.type_ = typ
+	er.numeric_ = u
+	er.collation_ = collationNumeric
+}
+
 func (er *EvalResult) setInt64(i int64) {
 	er.type_ = sqltypes.Int64
 	er.numeric_ = uint64(i)
@@ -296,7 +303,7 @@ func (er *EvalResult) Value() sqltypes.Value {
 	if er.expr != nil {
 		panic("did not resolve EvalResult after evaluation")
 	}
-	return er.toSQLValue(er.type_)
+	return er.value()
 }
 
 // TupleValues allows for retrieval of the value we expose for public consumption
@@ -344,7 +351,7 @@ func (er *EvalResult) ToBooleanStrict() (bool, error) {
 		return intToBool(er.uint64())
 	case sqltypes.Uint8, sqltypes.Uint16, sqltypes.Uint32, sqltypes.Uint64:
 		return intToBool(er.uint64())
-	case sqltypes.VarBinary:
+	case sqltypes.VarBinary, sqltypes.VarChar:
 		lower := strings.ToLower(er.string())
 		switch lower {
 		case "on":
@@ -355,7 +362,7 @@ func (er *EvalResult) ToBooleanStrict() (bool, error) {
 			return false, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "'%s' is not a boolean", lower)
 		}
 	}
-	return false, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "is not a boolean")
+	return false, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "'%s' is not a boolean", er.string())
 }
 
 func (er *EvalResult) textual() bool {
@@ -373,7 +380,7 @@ func (er *EvalResult) truthy() boolean {
 		return makeboolean(er.float64() != 0.0)
 	case sqltypes.Decimal:
 		return makeboolean(!er.decimal().num.IsZero())
-	case sqltypes.VarBinary:
+	case sqltypes.VarBinary, sqltypes.VarChar:
 		return makeboolean(parseStringToFloat(er.string()) != 0.0)
 	case querypb.Type_TUPLE:
 		panic("did not typecheck tuples")
@@ -382,6 +389,45 @@ func (er *EvalResult) truthy() boolean {
 	}
 }
 
+func formatMySQLFloat(typ querypb.Type, f float64) []byte {
+	format := byte('g')
+	if typ == sqltypes.Decimal {
+		format = 'f'
+	}
+
+	// the float printer in MySQL does not add a positive sign before
+	// the exponent for positive exponents, but the Golang printer does
+	// do that, and there's no way to customize it, so we must strip the
+	// redundant positive sign manually
+	// e.g. 1.234E+56789 -> 1.234E56789
+	fstr := strconv.AppendFloat(nil, f, format, -1, 64)
+	if idx := bytes.IndexByte(fstr, 'e'); idx >= 0 {
+		if fstr[idx+1] == '+' {
+			fstr = append(fstr[:idx+1], fstr[idx+2:]...)
+		}
+	}
+
+	return fstr
+}
+
+func (er *EvalResult) toRawBytes() []byte {
+	switch er.typeof() {
+	case sqltypes.Int64, sqltypes.Int32:
+		return strconv.AppendInt(nil, er.int64(), 10)
+	case sqltypes.Uint64, sqltypes.Uint32:
+		return strconv.AppendUint(nil, er.uint64(), 10)
+	case sqltypes.Float64, sqltypes.Float32:
+		return formatMySQLFloat(sqltypes.Float64, er.float64())
+	case sqltypes.Decimal:
+		dec := er.decimal()
+		return dec.num.FormatCustom(dec.frac, roundingModeFormat)
+	case sqltypes.Null:
+		return nil
+	default:
+		return er.bytes()
+	}
+}
+
 func (er *EvalResult) toSQLValue(resultType querypb.Type) sqltypes.Value {
 	switch {
 	case sqltypes.IsSigned(resultType):
@@ -407,11 +453,7 @@ func (er *EvalResult) toSQLValue(resultType querypb.Type) sqltypes.Value {
 		case sqltypes.Uint64, sqltypes.Uint32:
 			return sqltypes.MakeTrusted(resultType, strconv.AppendUint(nil, er.uint64(), 10))
 		case sqltypes.Float64, sqltypes.Float32:
-			format := byte('g')
-			if resultType == sqltypes.Decimal {
-				format = 'f'
-			}
-			return sqltypes.MakeTrusted(resultType, strconv.AppendFloat(nil, er.float64(), format, -1, 64))
+			return sqltypes.MakeTrusted(resultType, formatMySQLFloat(resultType, er.float64()))
 		case sqltypes.Decimal:
 			dec := er.decimal()
 			return sqltypes.MakeTrusted(resultType, dec.num.FormatCustom(dec.frac, roundingModeFormat))
@@ -575,8 +617,10 @@ func (er *EvalResult) setBindVar1(typ querypb.Type, value []byte, collation coll
 			throwEvalError(err)
 		}
 		er.setDecimal(dec)
-	case sqltypes.VarChar, sqltypes.Text, sqltypes.VarBinary:
-		er.setRaw(sqltypes.VarBinary, value, collation)
+	case sqltypes.VarChar, sqltypes.Text:
+		er.setRaw(sqltypes.VarChar, value, collation)
+	case sqltypes.VarBinary:
+		er.setRaw(sqltypes.VarBinary, value, collationBinary)
 	case sqltypes.Time, sqltypes.Datetime, sqltypes.Timestamp, sqltypes.Date:
 		er.setRaw(typ, value, collationNumeric)
 	case sqltypes.Null:
@@ -586,21 +630,6 @@ func (er *EvalResult) setBindVar1(typ querypb.Type, value []byte, collation coll
 	}
 }
 
-func (er *EvalResult) setBindVar(val *querypb.BindVariable, collation collations.TypedCollation) {
-	switch val.Type {
-	case querypb.Type_TUPLE:
-		tuple := make([]EvalResult, len(val.Values))
-		for i, value := range val.Values {
-			t := &tuple[i]
-			t.setBindVar1(value.Type, value.Value, collations.TypedCollation{})
-		}
-		er.setTuple(tuple)
-
-	default:
-		er.setBindVar1(val.Type, val.Value, collation)
-	}
-}
-
 // CoerceTo takes two input types, and decides how they should be coerced before compared
 func CoerceTo(v1, v2 querypb.Type) (querypb.Type, error) {
 	if v1 == v2 {
diff --git a/go/vt/vtgate/evalengine/expressions.go b/go/vt/vtgate/evalengine/expressions.go
index e66883f6e1..a61001ceb4 100644
--- a/go/vt/vtgate/evalengine/expressions.go
+++ b/go/vt/vtgate/evalengine/expressions.go
@@ -17,6 +17,7 @@ limitations under the License.
 package evalengine
 
 import (
+	"encoding/hex"
 	"strconv"
 	"unicode/utf8"
 
@@ -39,7 +40,6 @@ type (
 	Expr interface {
 		eval(env *ExpressionEnv, result *EvalResult)
 		typeof(env *ExpressionEnv) querypb.Type
-		collation() collations.TypedCollation
 		format(buf *formatter, depth int)
 		constant() bool
 		simplify() error
@@ -50,8 +50,9 @@ type (
 	}
 
 	BindVariable struct {
-		Key  string
-		coll collations.TypedCollation
+		Key        string
+		coll       collations.TypedCollation
+		coerceType querypb.Type
 	}
 
 	Column struct {
@@ -87,7 +88,7 @@ var _ Expr = (*CollateExpr)(nil)
 var _ Expr = (*LogicalExpr)(nil)
 var _ Expr = (*NotExpr)(nil)
 
-var noenv *ExpressionEnv = nil
+var noenv *ExpressionEnv
 
 type evalError struct {
 	error
@@ -106,9 +107,8 @@ func throwCardinalityError(expected int) {
 func (env *ExpressionEnv) cardinality(expr Expr) int {
 	switch expr := expr.(type) {
 	case *BindVariable:
-		val := expr.bvar(env)
-		if val.Type == querypb.Type_TUPLE {
-			return len(val.Values)
+		if expr.typeof(env) == querypb.Type_TUPLE {
+			return len(expr.bvar(env).Values)
 		}
 		return 1
 
@@ -129,7 +129,7 @@ func (env *ExpressionEnv) ensureCardinality(expr Expr, expected int) {
 func (env *ExpressionEnv) subexpr(expr Expr, nth int) (Expr, int) {
 	switch expr := expr.(type) {
 	case *BindVariable:
-		if expr.bvar(env).Type == querypb.Type_TUPLE {
+		if expr.typeof(env) == querypb.Type_TUPLE {
 			return nil, 1
 		}
 	case TupleExpr:
@@ -335,15 +335,26 @@ func NewLiteralString(val []byte, collation collations.TypedCollation) Expr {
 		}
 	}
 	lit := &Literal{}
-	lit.Val.setRaw(sqltypes.VarBinary, val, collation)
+	lit.Val.setRaw(sqltypes.VarChar, val, collation)
 	return lit
 }
 
+func NewLiteralBinaryFromHex(val []byte) (Expr, error) {
+	raw := make([]byte, hex.DecodedLen(len(val)))
+	if _, err := hex.Decode(raw, val); err != nil {
+		return nil, err
+	}
+	lit := &Literal{}
+	lit.Val.setRaw(sqltypes.VarBinary, raw, collationBinary)
+	return lit, nil
+}
+
 // NewBindVar returns a bind variable
 func NewBindVar(key string, collation collations.TypedCollation) Expr {
 	return &BindVariable{
-		Key:  key,
-		coll: collation,
+		Key:        key,
+		coll:       collation,
+		coerceType: -1,
 	}
 }
 
@@ -391,11 +402,30 @@ func (bv *BindVariable) bvar(env *ExpressionEnv) *querypb.BindVariable {
 
 // eval implements the Expr interface
 func (bv *BindVariable) eval(env *ExpressionEnv, result *EvalResult) {
-	result.setBindVar(bv.bvar(env), bv.coll)
+	bvar := bv.bvar(env)
+	typ := bvar.Type
+	if bv.coerceType >= 0 {
+		typ = bv.coerceType
+	}
+
+	switch typ {
+	case querypb.Type_TUPLE:
+		tuple := make([]EvalResult, len(bvar.Values))
+		for i, value := range bvar.Values {
+			tuple[i].setBindVar1(value.Type, value.Value, collations.TypedCollation{})
+		}
+		result.setTuple(tuple)
+
+	default:
+		result.setBindVar1(typ, bvar.Value, bv.coll)
+	}
 }
 
 // typeof implements the Expr interface
 func (bv *BindVariable) typeof(env *ExpressionEnv) querypb.Type {
+	if bv.coerceType >= 0 {
+		return bv.coerceType
+	}
 	return bv.bvar(env).Type
 }
 
diff --git a/go/vt/vtgate/evalengine/format.go b/go/vt/vtgate/evalengine/format.go
index 18daba9c2d..1e8209c29a 100644
--- a/go/vt/vtgate/evalengine/format.go
+++ b/go/vt/vtgate/evalengine/format.go
@@ -170,3 +170,20 @@ func (i *IsExpr) format(w *formatter, depth int) {
 		w.WriteString(" IS NOT FALSE")
 	}
 }
+
+func (c *CallExpression) format(w *formatter, depth int) {
+	w.Indent(depth)
+	w.WriteString(strings.ToUpper(c.Method))
+	w.WriteByte('(')
+	for i, expr := range c.Arguments {
+		if i > 0 {
+			w.WriteString(", ")
+		}
+		expr.format(w, depth+1)
+		if !c.Aliases[i].IsEmpty() {
+			w.WriteString(" AS ")
+			w.WriteString(c.Aliases[i].String())
+		}
+	}
+	w.WriteByte(')')
+}
diff --git a/go/vt/vtgate/evalengine/func.go b/go/vt/vtgate/evalengine/func.go
new file mode 100644
index 0000000000..91eb2ad821
--- /dev/null
+++ b/go/vt/vtgate/evalengine/func.go
@@ -0,0 +1,241 @@
+/*
+Copyright 2021 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 evalengine
+
+import (
+	"bytes"
+	"fmt"
+	"math"
+
+	"vitess.io/vitess/go/mysql/collations"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+	"vitess.io/vitess/go/vt/sqlparser"
+)
+
+type CallExpression struct {
+	Arguments TupleExpr
+	Aliases   []sqlparser.ColIdent
+	Method    string
+	Call      func([]EvalResult, *EvalResult)
+}
+
+func (c *CallExpression) typeof(*ExpressionEnv) querypb.Type {
+	return -1
+}
+
+func (c *CallExpression) constant() bool {
+	return c.Arguments.constant()
+}
+
+func (c *CallExpression) simplify() error {
+	return c.Arguments.simplify()
+}
+
+var _ Expr = (*CallExpression)(nil)
+
+func (c *CallExpression) eval(env *ExpressionEnv, result *EvalResult) {
+	var args []EvalResult = make([]EvalResult, len(c.Arguments))
+	for i, arg := range c.Arguments {
+		args[i].init(env, arg)
+	}
+	c.Call(args, result)
+}
+
+func builtinFuncCoalesce(args []EvalResult, result *EvalResult) {
+	for _, arg := range args {
+		if !arg.null() {
+			*result = arg
+			result.resolve()
+			return
+		}
+	}
+	result.setNull()
+}
+
+type multiComparisonFunc func(args []EvalResult, result *EvalResult, cmp int)
+
+func getMultiComparisonFunc(args []EvalResult) multiComparisonFunc {
+	var (
+		integers int
+		floats   int
+		decimals int
+		text     int
+		binary   int
+	)
+
+	/*
+		If any argument is NULL, the result is NULL. No comparison is needed.
+		If all arguments are integer-valued, they are compared as integers.
+		If at least one argument is double precision, they are compared as double-precision values. Otherwise, if at least one argument is a DECIMAL value, they are compared as DECIMAL values.
+		If the arguments comprise a mix of numbers and strings, they are compared as strings.
+		If any argument is a nonbinary (character) string, the arguments are compared as nonbinary strings.
+		In all other cases, the arguments are compared as binary strings.
+	*/
+
+	for _, arg := range args {
+		switch arg.typeof() {
+		case querypb.Type_NULL_TYPE:
+			return func(args []EvalResult, result *EvalResult, cmp int) {
+				result.setNull()
+			}
+		case querypb.Type_INT8, querypb.Type_INT16, querypb.Type_INT32, querypb.Type_INT64:
+			integers++
+		case querypb.Type_UINT8, querypb.Type_UINT16, querypb.Type_UINT32, querypb.Type_UINT64:
+			if arg.uint64() > math.MaxInt64 {
+				decimals++
+			} else {
+				integers++
+			}
+		case querypb.Type_FLOAT32, querypb.Type_FLOAT64:
+			floats++
+		case querypb.Type_DECIMAL:
+			decimals++
+		case querypb.Type_TEXT, querypb.Type_VARCHAR:
+			text++
+		case querypb.Type_BLOB, querypb.Type_BINARY, querypb.Type_VARBINARY:
+			binary++
+		}
+	}
+
+	if integers == len(args) {
+		return compareAllInteger
+	}
+	if binary > 0 || text > 0 {
+		if binary > 0 {
+			return compareAllBinary
+		}
+		if text > 0 {
+			return compareAllText
+		}
+	} else {
+		if floats > 0 {
+			return compareAllFloat
+		}
+		if decimals > 0 {
+			return compareAllDecimal
+		}
+	}
+	panic("unexpected argument type")
+}
+
+func compareAllInteger(args []EvalResult, result *EvalResult, cmp int) {
+	var candidateI = args[0].int64()
+	for _, arg := range args[1:] {
+		thisI := arg.int64()
+		if (cmp < 0) == (thisI < candidateI) {
+			candidateI = thisI
+		}
+	}
+	result.setInt64(candidateI)
+}
+
+func compareAllFloat(args []EvalResult, result *EvalResult, cmp int) {
+	candidateF, err := args[0].coerceToFloat()
+	if err != nil {
+		throwEvalError(err)
+	}
+
+	for _, arg := range args[1:] {
+		thisF, err := arg.coerceToFloat()
+		if err != nil {
+			throwEvalError(err)
+		}
+		if (cmp < 0) == (thisF < candidateF) {
+			candidateF = thisF
+		}
+	}
+	result.setFloat(candidateF)
+}
+
+func compareAllDecimal(args []EvalResult, result *EvalResult, cmp int) {
+	candidateD := args[0].coerceToDecimal()
+	maxFrac := candidateD.frac
+
+	for _, arg := range args[1:] {
+		thisD := arg.coerceToDecimal()
+		if (cmp < 0) == (thisD.num.Cmp(&candidateD.num) < 0) {
+			candidateD = thisD
+		}
+		if thisD.frac > maxFrac {
+			maxFrac = thisD.frac
+		}
+	}
+
+	candidateD.frac = maxFrac
+	result.setDecimal(candidateD)
+}
+
+func compareAllText(args []EvalResult, result *EvalResult, cmp int) {
+	env := collations.Local()
+	candidateB := args[0].toRawBytes()
+	collationB := args[0].collation()
+
+	for _, arg := range args[1:] {
+		thisB := arg.toRawBytes()
+		thisTC, coerceLeft, coerceRight, err := env.MergeCollations(arg.collation(), collationB, collations.CoercionOptions{ConvertToSuperset: true, ConvertWithCoercion: true})
+		if err != nil {
+			throwEvalError(err)
+		}
+
+		collation := env.LookupByID(thisTC.Collation)
+
+		var leftB = thisB
+		var rightB = candidateB
+		if coerceLeft != nil {
+			leftB, _ = coerceLeft(nil, leftB)
+		}
+		if coerceRight != nil {
+			rightB, _ = coerceRight(nil, rightB)
+		}
+		if (cmp < 0) == (collation.Collate(leftB, rightB, false) < 0) {
+			candidateB = thisB
+		}
+	}
+
+	result.setRaw(querypb.Type_VARCHAR, candidateB, collationB)
+}
+
+func compareAllBinary(args []EvalResult, result *EvalResult, cmp int) {
+	candidateB := args[0].toRawBytes()
+
+	for _, arg := range args[1:] {
+		thisB := arg.toRawBytes()
+		if (cmp < 0) == (bytes.Compare(thisB, candidateB) < 0) {
+			candidateB = thisB
+		}
+	}
+
+	result.setRaw(querypb.Type_VARBINARY, candidateB, collationBinary)
+}
+
+func throwArgError(fname string) {
+	panic(evalError{fmt.Errorf("Incorrect parameter count in the call to native function '%s'", fname)})
+}
+
+func builtinFuncLeast(args []EvalResult, result *EvalResult) {
+	if len(args) < 2 {
+		throwArgError("LEAST")
+	}
+	getMultiComparisonFunc(args)(args, result, -1)
+}
+
+func builtinFuncGreatest(args []EvalResult, result *EvalResult) {
+	if len(args) < 2 {
+		throwArgError("GREATEST")
+	}
+	getMultiComparisonFunc(args)(args, result, 1)
+}
diff --git a/go/vt/vtgate/evalengine/logical.go b/go/vt/vtgate/evalengine/logical.go
index bd30533813..762136dfc1 100644
--- a/go/vt/vtgate/evalengine/logical.go
+++ b/go/vt/vtgate/evalengine/logical.go
@@ -17,7 +17,6 @@ limitations under the License.
 package evalengine
 
 import (
-	"vitess.io/vitess/go/mysql/collations"
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	"vitess.io/vitess/go/vt/sqlparser"
 )
@@ -62,14 +61,14 @@ func makeboolean2(b, isNull bool) boolean {
 	return makeboolean(b)
 }
 
-func (b boolean) not() boolean {
-	switch b {
+func (left boolean) not() boolean {
+	switch left {
 	case boolFalse:
 		return boolTrue
 	case boolTrue:
 		return boolFalse
 	default:
-		return b
+		return left
 	}
 }
 
@@ -135,10 +134,6 @@ func (n *NotExpr) typeof(*ExpressionEnv) querypb.Type {
 	return querypb.Type_UINT64
 }
 
-func (n *NotExpr) collation() collations.TypedCollation {
-	return collationNumeric
-}
-
 func (l *LogicalExpr) eval(env *ExpressionEnv, out *EvalResult) {
 	var left, right EvalResult
 	left.init(env, l.Left)
@@ -153,10 +148,6 @@ func (l *LogicalExpr) typeof(env *ExpressionEnv) querypb.Type {
 	return querypb.Type_UINT64
 }
 
-func (n *LogicalExpr) collation() collations.TypedCollation {
-	return collationNumeric
-}
-
 // IsExpr represents the IS expression in MySQL.
 // boolean_primary IS [NOT] {TRUE | FALSE | NULL}
 type IsExpr struct {
@@ -176,7 +167,3 @@ func (i *IsExpr) eval(env *ExpressionEnv, result *EvalResult) {
 func (i *IsExpr) typeof(env *ExpressionEnv) querypb.Type {
 	return querypb.Type_INT64
 }
-
-func (i *IsExpr) collation() collations.TypedCollation {
-	return collationNumeric
-}
diff --git a/go/vt/vtgate/evalengine/simplify.go b/go/vt/vtgate/evalengine/simplify.go
index aafb5033d1..cc63da3d22 100644
--- a/go/vt/vtgate/evalengine/simplify.go
+++ b/go/vt/vtgate/evalengine/simplify.go
@@ -76,13 +76,14 @@ func (expr *BinaryExpr) simplify() error {
 }
 
 func (expr *LikeExpr) simplify() error {
-	if err := expr.BinaryCoercedExpr.simplify(); err != nil {
+	if err := expr.BinaryExpr.simplify(); err != nil {
 		return err
 	}
 
 	lit2, _ := expr.Right.(*Literal)
-	if lit2 != nil && lit2.Val.textual() && expr.MergedCollation.Valid() {
-		coll := collations.Local().LookupByID(expr.MergedCollation.Collation)
+	if lit2 != nil && lit2.Val.textual() {
+		expr.MatchCollation = lit2.Val.collation().Collation
+		coll := collations.Local().LookupByID(expr.MatchCollation)
 		expr.Match = coll.Wildcard(lit2.Val.bytes(), 0, 0, 0)
 	}
 	return nil
@@ -144,30 +145,6 @@ func (inexpr *InExpr) simplify() error {
 	return nil
 }
 
-func (expr *BinaryCoercedExpr) simplify() error {
-	var err error
-
-	if err = expr.BinaryExpr.simplify(); err != nil {
-		return err
-	}
-
-	lit1, _ := expr.Left.(*Literal)
-	lit2, _ := expr.Right.(*Literal)
-
-	if lit1 != nil && expr.CoerceLeft != nil {
-		b, _ := expr.CoerceLeft(nil, lit1.Val.bytes())
-		lit1.Val.replaceBytes(b, expr.MergedCollation)
-		expr.CoerceLeft = nil
-	}
-	if lit2 != nil && expr.CoerceRight != nil {
-		b, _ := expr.CoerceRight(nil, lit2.Val.bytes())
-		lit2.Val.replaceBytes(b, expr.MergedCollation)
-		expr.CoerceRight = nil
-	}
-
-	return nil
-}
-
 func (expr TupleExpr) simplify() error {
 	var err error
 	for i, subexpr := range expr {
diff --git a/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go b/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
index eb3e4b241d..53e41e64cf 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
@@ -160,15 +160,14 @@ func getOpcode(comparison *sqlparser.ComparisonExpr) (Opcode, error) {
 }
 
 // compare returns true after applying the comparison specified in the Filter to the actual data in the column
-func compare(comparison Opcode, columnValue, filterValue sqltypes.Value) (bool, error) {
+func compare(comparison Opcode, columnValue, filterValue sqltypes.Value, charset collations.ID) (bool, error) {
 	// use null semantics: return false if either value is null
 	if columnValue.IsNull() || filterValue.IsNull() {
 		return false, nil
 	}
 	// at this point neither values can be null
 	// NullsafeCompare returns 0 if values match, -1 if columnValue < filterValue, 1 if columnValue > filterValue
-	// TODO(king-11) make collation aware
-	result, err := evalengine.NullsafeCompare(columnValue, filterValue, collations.Unknown)
+	result, err := evalengine.NullsafeCompare(columnValue, filterValue, charset)
 	if err != nil {
 		return false, err
 	}
@@ -208,7 +207,7 @@ func compare(comparison Opcode, columnValue, filterValue sqltypes.Value) (bool,
 // The output of the filtering operation is stored in the 'result' argument because
 // filtering cannot be performed in-place. The result argument must be a slice of
 // length equal to ColExprs
-func (plan *Plan) filter(values, result []sqltypes.Value) (bool, error) {
+func (plan *Plan) filter(values, result []sqltypes.Value, charsets []collations.ID) (bool, error) {
 	if len(result) != len(plan.ColExprs) {
 		return false, fmt.Errorf("expected %d values in result slice", len(plan.ColExprs))
 	}
@@ -223,7 +222,7 @@ func (plan *Plan) filter(values, result []sqltypes.Value) (bool, error) {
 				return false, nil
 			}
 		default:
-			match, err := compare(filter.Opcode, values[filter.ColNum], filter.Value)
+			match, err := compare(filter.Opcode, values[filter.ColNum], filter.Value, charsets[filter.ColNum])
 			if err != nil {
 				return false, err
 			}
diff --git a/go/vt/vttablet/tabletserver/vstreamer/rowstreamer.go b/go/vt/vttablet/tabletserver/vstreamer/rowstreamer.go
index 8cb0b8696b..b74be71f9c 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/rowstreamer.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/rowstreamer.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"time"
 
+	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/textutil"
 	"vitess.io/vitess/go/vt/dbconfigs"
@@ -260,6 +261,11 @@ func (rs *rowStreamer) streamQuery(conn *snapshotConn, send func(*binlogdatapb.V
 		}
 	}
 
+	charsets := make([]collations.ID, len(flds))
+	for i, fld := range flds {
+		charsets[i] = collations.ID(fld.Charset)
+	}
+
 	err = send(&binlogdatapb.VStreamRowsResponse{
 		Fields:   rs.plan.fields(),
 		Pkfields: pkfields,
@@ -305,7 +311,7 @@ func (rs *rowStreamer) streamQuery(conn *snapshotConn, send func(*binlogdatapb.V
 			lastpk[i] = mysqlrow[pk]
 		}
 		// Reuse the vstreamer's filter.
-		ok, err := rs.plan.filter(mysqlrow, filtered)
+		ok, err := rs.plan.filter(mysqlrow, filtered, charsets)
 		if err != nil {
 			return err
 		}
diff --git a/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go b/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go
index 2f63efa514..5a61041ba6 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go
@@ -25,6 +25,7 @@ import (
 
 	"google.golang.org/protobuf/encoding/prototext"
 
+	"vitess.io/vitess/go/mysql/collations"
 	vtschema "vitess.io/vitess/go/vt/schema"
 
 	"vitess.io/vitess/go/mysql"
@@ -884,6 +885,7 @@ func (vs *vstreamer) extractRowAndFilter(plan *streamerPlan, data []byte, dataCo
 		return false, nil, nil
 	}
 	values := make([]sqltypes.Value, dataColumns.Count())
+	charsets := make([]collations.ID, len(values))
 	valueIndex := 0
 	pos := 0
 	for colNum := 0; colNum < dataColumns.Count(); colNum++ {
@@ -902,11 +904,12 @@ func (vs *vstreamer) extractRowAndFilter(plan *streamerPlan, data []byte, dataCo
 		}
 		pos += l
 
+		charsets[colNum] = collations.ID(plan.Table.Fields[colNum].Charset)
 		values[colNum] = value
 		valueIndex++
 	}
 	filtered := make([]sqltypes.Value, len(plan.ColExprs))
-	ok, err := plan.filter(values, filtered)
+	ok, err := plan.filter(values, filtered, charsets)
 	return ok, filtered, err
 }
 


Test patch:
diff --git a/go/mysql/collations/integration/main_test.go b/go/mysql/collations/integration/main_test.go
index 4c524bec60..16791c1983 100644
--- a/go/mysql/collations/integration/main_test.go
+++ b/go/mysql/collations/integration/main_test.go
@@ -71,6 +71,7 @@ func TestMain(m *testing.M) {
 				},
 			},
 			OnlyMySQL: true,
+			Charset:   "utf8mb4",
 		}
 		cluster := vttest.LocalCluster{
 			Config: cfg,
diff --git a/go/sqltypes/bind_variables_test.go b/go/sqltypes/bind_variables_test.go
index fdf34755b6..2d975b52e5 100644
--- a/go/sqltypes/bind_variables_test.go
+++ b/go/sqltypes/bind_variables_test.go
@@ -88,7 +88,7 @@ func TestBuildBindVariable(t *testing.T) {
 	}{{
 		in: "aa",
 		out: &querypb.BindVariable{
-			Type:  querypb.Type_VARBINARY,
+			Type:  querypb.Type_VARCHAR,
 			Value: []byte("aa"),
 		},
 	}, {
@@ -156,7 +156,7 @@ func TestBuildBindVariable(t *testing.T) {
 		out: &querypb.BindVariable{
 			Type: querypb.Type_TUPLE,
 			Values: []*querypb.Value{{
-				Type:  querypb.Type_VARBINARY,
+				Type:  querypb.Type_VARCHAR,
 				Value: []byte("aa"),
 			}, {
 				Type:  querypb.Type_INT64,
@@ -168,10 +168,10 @@ func TestBuildBindVariable(t *testing.T) {
 		out: &querypb.BindVariable{
 			Type: querypb.Type_TUPLE,
 			Values: []*querypb.Value{{
-				Type:  querypb.Type_VARBINARY,
+				Type:  querypb.Type_VARCHAR,
 				Value: []byte("aa"),
 			}, {
-				Type:  querypb.Type_VARBINARY,
+				Type:  querypb.Type_VARCHAR,
 				Value: []byte("bb"),
 			}},
 		},
@@ -606,7 +606,7 @@ func TestBindVariablesFormat(t *testing.T) {
 
 	formattedStr = FormatBindVariables(bindVariables, true /* full */, true /* asJSON */)
 	t.Logf("%q", formattedStr)
-	if !strings.Contains(formattedStr, "\"key_1\": {\"type\": \"VARBINARY\", \"value\": \"val_1\"}") {
+	if !strings.Contains(formattedStr, "\"key_1\": {\"type\": \"VARCHAR\", \"value\": \"val_1\"}") {
 		t.Fatalf("bind variable 'key_1' is not formatted")
 	}
 
@@ -623,7 +623,7 @@ func TestBindVariablesFormat(t *testing.T) {
 	}
 
 	formattedStr = FormatBindVariables(bindVariables, false /* full */, true /* asJSON */)
-	if !strings.Contains(formattedStr, "\"key_1\": {\"type\": \"VARBINARY\", \"value\": \"5 bytes\"}") {
+	if !strings.Contains(formattedStr, "\"key_1\": {\"type\": \"VARCHAR\", \"value\": \"5 bytes\"}") {
 		t.Fatalf("bind variable 'key_1' is not formatted")
 	}
 
@@ -631,11 +631,11 @@ func TestBindVariablesFormat(t *testing.T) {
 		t.Fatalf("bind variable 'key_2' is not formatted")
 	}
 
-	if !strings.Contains(formattedStr, "\"key_3\": {\"type\": \"VARBINARY\", \"value\": \"5 bytes\"}") {
+	if !strings.Contains(formattedStr, "\"key_3\": {\"type\": \"VARCHAR\", \"value\": \"5 bytes\"}") {
 		t.Fatalf("bind variable 'key_3' is not formatted")
 	}
 
-	if !strings.Contains(formattedStr, "\"key_4\": {\"type\": \"VARBINARY\", \"value\": \"2 items\"}") {
+	if !strings.Contains(formattedStr, "\"key_4\": {\"type\": \"VARCHAR\", \"value\": \"2 items\"}") {
 		t.Fatalf("bind variable 'key_4' is not formatted")
 	}
 }
diff --git a/go/test/endtoend/vtgate/misc_test.go b/go/test/endtoend/vtgate/misc_test.go
index 8b37d39706..6d846db78f 100644
--- a/go/test/endtoend/vtgate/misc_test.go
+++ b/go/test/endtoend/vtgate/misc_test.go
@@ -285,15 +285,15 @@ func TestSwitchBetweenOlapAndOltp(t *testing.T) {
 	require.NoError(t, err)
 	defer conn.Close()
 
-	utils.AssertMatches(t, conn, "select @@workload", `[[VARBINARY("OLTP")]]`)
+	utils.AssertMatches(t, conn, "select @@workload", `[[VARCHAR("OLTP")]]`)
 
 	utils.Exec(t, conn, "set workload='olap'")
 
-	utils.AssertMatches(t, conn, "select @@workload", `[[VARBINARY("OLAP")]]`)
+	utils.AssertMatches(t, conn, "select @@workload", `[[VARCHAR("OLAP")]]`)
 
 	utils.Exec(t, conn, "set workload='oltp'")
 
-	utils.AssertMatches(t, conn, "select @@workload", `[[VARBINARY("OLTP")]]`)
+	utils.AssertMatches(t, conn, "select @@workload", `[[VARCHAR("OLTP")]]`)
 }
 
 func TestFoundRowsOnDualQueries(t *testing.T) {
@@ -528,7 +528,7 @@ func TestRenameFieldsOnOLAP(t *testing.T) {
 	assert.Equal(t, `Tables_in_ks`, fmt.Sprintf("%v", qr.Fields[0].Name))
 	_ = utils.Exec(t, conn, "use mysql")
 	qr = utils.Exec(t, conn, "select @@workload")
-	assert.Equal(t, `[[VARBINARY("OLAP")]]`, fmt.Sprintf("%v", qr.Rows))
+	assert.Equal(t, `[[VARCHAR("OLAP")]]`, fmt.Sprintf("%v", qr.Rows))
 }
 
 func TestSelectEqualUniqueOuterJoinRightPredicate(t *testing.T) {
diff --git a/go/test/endtoend/vtgate/readafterwrite/raw_test.go b/go/test/endtoend/vtgate/readafterwrite/raw_test.go
index 01b5ac72fa..395c0d3c57 100644
--- a/go/test/endtoend/vtgate/readafterwrite/raw_test.go
+++ b/go/test/endtoend/vtgate/readafterwrite/raw_test.go
@@ -147,9 +147,9 @@ func TestRAWSettings(t *testing.T) {
 	require.NoError(t, err)
 	defer conn.Close()
 
-	assertMatches(t, conn, `select @@read_after_write_gtid, @@read_after_write_timeout, @@session_track_gtids`, `[[VARBINARY("") FLOAT64(0) VARBINARY("off")]]`)
+	assertMatches(t, conn, `select @@read_after_write_gtid, @@read_after_write_timeout, @@session_track_gtids`, `[[VARCHAR("") FLOAT64(0) VARCHAR("off")]]`)
 	exec(t, conn, `set read_after_write_gtid = 'some-gtid:1', read_after_write_timeout = 0.2, session_track_gtids = own_gtid`)
-	assertMatches(t, conn, `select @@read_after_write_gtid, @@read_after_write_timeout, @@session_track_gtids`, `[[VARBINARY("some-gtid:1") FLOAT64(0.2) VARBINARY("own_gtid")]]`)
+	assertMatches(t, conn, `select @@read_after_write_gtid, @@read_after_write_timeout, @@session_track_gtids`, `[[VARCHAR("some-gtid:1") FLOAT64(0.2) VARCHAR("own_gtid")]]`)
 }
 
 func exec(t *testing.T, conn *mysql.Conn, query string) (*sqltypes.Result, error) {
diff --git a/go/test/endtoend/vtgate/reservedconn/udv_test.go b/go/test/endtoend/vtgate/reservedconn/udv_test.go
index 5c2efa4849..34d386179a 100644
--- a/go/test/endtoend/vtgate/reservedconn/udv_test.go
+++ b/go/test/endtoend/vtgate/reservedconn/udv_test.go
@@ -55,7 +55,7 @@ func TestSetUDV(t *testing.T) {
 		query: "/*!40101 SET @OLD_SQL_MODE=@@SQL_MODE */",
 	}, { // This is handled at vtgate.
 		query:        "select @foo, @bar, @baz, @tablet",
-		expectedRows: `[[VARBINARY("abc") INT64(42) DECIMAL(30.5) VARBINARY("foobar")]]`, rowsReturned: 1,
+		expectedRows: `[[VARCHAR("abc") INT64(42) DECIMAL(30.5) VARBINARY("foobar")]]`, rowsReturned: 1,
 	}, { // Cannot really check a specific value for sql_mode as it will differ based on database selected to run these tests.
 		query:        "select @OLD_SQL_MODE = @@SQL_MODE",
 		expectedRows: `[[INT64(1)]]`, rowsReturned: 1,
diff --git a/go/vt/sqlparser/normalizer_test.go b/go/vt/sqlparser/normalizer_test.go
index e60ed3b92d..dc76d2f898 100644
--- a/go/vt/sqlparser/normalizer_test.go
+++ b/go/vt/sqlparser/normalizer_test.go
@@ -42,7 +42,7 @@ func TestNormalize(t *testing.T) {
 		in:      "select * from t where v1 = 'aa'",
 		outstmt: "select * from t where v1 = :bv1",
 		outbv: map[string]*querypb.BindVariable{
-			"bv1": sqltypes.BytesBindVariable([]byte("aa")),
+			"bv1": sqltypes.StringBindVariable("aa"),
 		},
 	}, {
 		// placeholder
@@ -59,7 +59,7 @@ func TestNormalize(t *testing.T) {
 		in:      "select 'aa' from t",
 		outstmt: "select :bv1 from t",
 		outbv: map[string]*querypb.BindVariable{
-			"bv1": sqltypes.BytesBindVariable([]byte("aa")),
+			"bv1": sqltypes.StringBindVariable("aa"),
 		},
 	}, {
 		// int val
@@ -103,7 +103,7 @@ func TestNormalize(t *testing.T) {
 		outstmt: "select * from t where v1 = :bv1 and v2 = :bv2",
 		outbv: map[string]*querypb.BindVariable{
 			"bv1": sqltypes.Int64BindVariable(1),
-			"bv2": sqltypes.BytesBindVariable([]byte("1")),
+			"bv2": sqltypes.StringBindVariable("1"),
 		},
 	}, {
 		// val should not be reused for non-select statements
@@ -183,15 +183,15 @@ func TestNormalize(t *testing.T) {
 		in:      fmt.Sprintf("select * from t where v1 = '%256s' and v2 = '%256s'", "a", "a"),
 		outstmt: "select * from t where v1 = :bv1 and v2 = :bv1",
 		outbv: map[string]*querypb.BindVariable{
-			"bv1": sqltypes.BytesBindVariable([]byte(fmt.Sprintf("%256s", "a"))),
+			"bv1": sqltypes.StringBindVariable(fmt.Sprintf("%256s", "a")),
 		},
 	}, {
 		// Values greater than len 256 will not reuse.
 		in:      fmt.Sprintf("select * from t where v1 = '%257s' and v2 = '%257s'", "b", "b"),
 		outstmt: "select * from t where v1 = :bv1 and v2 = :bv2",
 		outbv: map[string]*querypb.BindVariable{
-			"bv1": sqltypes.BytesBindVariable([]byte(fmt.Sprintf("%257s", "b"))),
-			"bv2": sqltypes.BytesBindVariable([]byte(fmt.Sprintf("%257s", "b"))),
+			"bv1": sqltypes.StringBindVariable(fmt.Sprintf("%257s", "b")),
+			"bv2": sqltypes.StringBindVariable(fmt.Sprintf("%257s", "b")),
 		},
 	}, {
 		// bad int
@@ -220,14 +220,14 @@ func TestNormalize(t *testing.T) {
 		in:      "select * from t where v1 in (1, '2')",
 		outstmt: "select * from t where v1 in ::bv1",
 		outbv: map[string]*querypb.BindVariable{
-			"bv1": sqltypes.TestBindVariable([]interface{}{1, []byte("2")}),
+			"bv1": sqltypes.TestBindVariable([]interface{}{1, "2"}),
 		},
 	}, {
 		// NOT IN clause
 		in:      "select * from t where v1 not in (1, '2')",
 		outstmt: "select * from t where v1 not in ::bv1",
 		outbv: map[string]*querypb.BindVariable{
-			"bv1": sqltypes.TestBindVariable([]interface{}{1, []byte("2")}),
+			"bv1": sqltypes.TestBindVariable([]interface{}{1, "2"}),
 		},
 	}, {
 		// Do not normalize cast/convert types
diff --git a/go/vt/vtgate/autocommit_test.go b/go/vt/vtgate/autocommit_test.go
index 9ccc4243a6..2f9a14e660 100644
--- a/go/vt/vtgate/autocommit_test.go
+++ b/go/vt/vtgate/autocommit_test.go
@@ -17,12 +17,11 @@ limitations under the License.
 package vtgate
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 
-	"context"
-
 	"vitess.io/vitess/go/sqltypes"
 
 	querypb "vitess.io/vitess/go/vt/proto/query"
@@ -95,15 +94,15 @@ func TestAutocommitUpdateVindexChange(t *testing.T) {
 	assertQueries(t, sbclookup, []*querypb.BoundQuery{{
 		Sql: "delete from name_lastname_keyspace_id_map where `name` = :name and lastname = :lastname and keyspace_id = :keyspace_id",
 		BindVariables: map[string]*querypb.BindVariable{
-			"lastname":    sqltypes.ValueBindVariable(sqltypes.NewVarChar("foo")),
+			"lastname":    sqltypes.StringBindVariable("foo"),
 			"name":        sqltypes.Int32BindVariable(1),
 			"keyspace_id": sqltypes.BytesBindVariable([]byte("\x16k@\xb4J\xbaK\xd6")),
 		},
 	}, {
 		Sql: "insert into name_lastname_keyspace_id_map(`name`, lastname, keyspace_id) values (:name_0, :lastname_0, :keyspace_id_0)",
 		BindVariables: map[string]*querypb.BindVariable{
-			"name_0":        sqltypes.BytesBindVariable([]byte("myname")),
-			"lastname_0":    sqltypes.BytesBindVariable([]byte("mylastname")),
+			"name_0":        sqltypes.StringBindVariable("myname"),
+			"lastname_0":    sqltypes.StringBindVariable("mylastname"),
 			"keyspace_id_0": sqltypes.BytesBindVariable([]byte("\x16k@\xb4J\xbaK\xd6")),
 		},
 	}})
@@ -264,7 +263,7 @@ func TestAutocommitInsertLookup(t *testing.T) {
 	assertQueries(t, sbclookup, []*querypb.BoundQuery{{
 		Sql: "insert into name_user_map(`name`, user_id) values (:name_0, :user_id_0)",
 		BindVariables: map[string]*querypb.BindVariable{
-			"name_0":    sqltypes.BytesBindVariable([]byte("myname")),
+			"name_0":    sqltypes.StringBindVariable("myname"),
 			"user_id_0": sqltypes.Uint64BindVariable(1),
 		},
 	}})
@@ -274,7 +273,7 @@ func TestAutocommitInsertLookup(t *testing.T) {
 		Sql: "insert into `user`(id, v, `name`) values (:_Id_0, 2, :_name_0)",
 		BindVariables: map[string]*querypb.BindVariable{
 			"_Id_0":   sqltypes.Int64BindVariable(1),
-			"_name_0": sqltypes.BytesBindVariable([]byte("myname")),
+			"_name_0": sqltypes.StringBindVariable("myname"),
 			"__seq0":  sqltypes.Int64BindVariable(1),
 		},
 	}})
diff --git a/go/vt/vtgate/endtoend/misc_test.go b/go/vt/vtgate/endtoend/misc_test.go
index 9cc2898632..138b68d0aa 100644
--- a/go/vt/vtgate/endtoend/misc_test.go
+++ b/go/vt/vtgate/endtoend/misc_test.go
@@ -36,7 +36,7 @@ func TestDatabaseFunc(t *testing.T) {
 
 	exec(t, conn, "use ks")
 	qr := exec(t, conn, "select database()")
-	require.Equal(t, `[[VARBINARY("ks")]]`, fmt.Sprintf("%v", qr.Rows))
+	require.Equal(t, `[[VARCHAR("ks")]]`, fmt.Sprintf("%v", qr.Rows))
 }
 
 func TestSysNumericPrecisionScale(t *testing.T) {
diff --git a/go/vt/vtgate/engine/route_test.go b/go/vt/vtgate/engine/route_test.go
index 99e24eb68a..3aff82b3f4 100644
--- a/go/vt/vtgate/engine/route_test.go
+++ b/go/vt/vtgate/engine/route_test.go
@@ -105,7 +105,7 @@ func TestInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T) {
 		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\" table_name: type:VARCHAR value:\"routedTable\"} false false"},
 	}, {
 		testName:    "both schema and table predicates - not routed",
 		tableSchema: []string{"schema"},
@@ -114,7 +114,7 @@ func TestInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T) {
 		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\" table_name: type:VARCHAR value:\"table\"} false false"},
 	}, {
 		testName:    "multiple schema and table predicates",
 		tableSchema: []string{"schema", "schema", "schema"},
@@ -125,7 +125,7 @@ func TestInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T) {
 			"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\" t1: type:VARCHAR value:\"table\" t2: type:VARCHAR value:\"table\" t3: type:VARCHAR value:\"table\"} false false"},
 	}, {
 		testName:  "table name predicate - routed table",
 		tableName: map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("tableName"), collations.TypedCollation{})},
@@ -133,7 +133,7 @@ func TestInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T) {
 		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 {table_name: type:VARCHAR value:\"routedTable\"} false false"},
 	}, {
 		testName:  "table name predicate - not routed",
 		tableName: map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("tableName"), collations.TypedCollation{})},
@@ -141,7 +141,7 @@ func TestInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T) {
 		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 {table_name: type:VARCHAR value:\"tableName\"} false false"},
 	}, {
 		testName:    "schema predicate",
 		tableSchema: []string{"myKeyspace"},
@@ -616,7 +616,7 @@ func TestSelectLike(t *testing.T) {
 	// range 0c-0d hits 2 shards ks.-0c80 ks.0c80-0d;
 	// note that 0c-0d should not hit ks.0d-40
 	vc.ExpectLog(t, []string{
-		`ResolveDestinations ks [type:VARBINARY value:"a%"] Destinations:DestinationKeyRange(0c-0d)`,
+		`ResolveDestinations ks [type:VARCHAR value:"a%"] Destinations:DestinationKeyRange(0c-0d)`,
 		`ExecuteMultiShard ks.-0c80: dummy_select {} ks.0c80-0d: dummy_select {} false false`,
 	})
 	expectResult(t, "sel.Execute", result, defaultSelectResult)
@@ -627,7 +627,7 @@ func TestSelectLike(t *testing.T) {
 	require.NoError(t, err)
 
 	vc.ExpectLog(t, []string{
-		`ResolveDestinations ks [type:VARBINARY value:"a%"] Destinations:DestinationKeyRange(0c-0d)`,
+		`ResolveDestinations ks [type:VARCHAR value:"a%"] Destinations:DestinationKeyRange(0c-0d)`,
 		`StreamExecuteMulti dummy_select ks.-0c80: {} ks.0c80-0d: {} `,
 	})
 	expectResult(t, "sel.StreamExecute", result, defaultSelectResult)
@@ -646,7 +646,7 @@ func TestSelectLike(t *testing.T) {
 		t.Fatal(err)
 	}
 	vc.ExpectLog(t, []string{
-		`ResolveDestinations ks [type:VARBINARY value:"ab%"] Destinations:DestinationKeyRange(0c92-0c93)`,
+		`ResolveDestinations ks [type:VARCHAR value:"ab%"] Destinations:DestinationKeyRange(0c92-0c93)`,
 		`ExecuteMultiShard ks.0c80-0d: dummy_select {} false false`,
 	})
 	expectResult(t, "sel.Execute", result, defaultSelectResult)
@@ -657,7 +657,7 @@ func TestSelectLike(t *testing.T) {
 	require.NoError(t, err)
 
 	vc.ExpectLog(t, []string{
-		`ResolveDestinations ks [type:VARBINARY value:"ab%"] Destinations:DestinationKeyRange(0c92-0c93)`,
+		`ResolveDestinations ks [type:VARCHAR value:"ab%"] Destinations:DestinationKeyRange(0c92-0c93)`,
 		`StreamExecuteMulti dummy_select ks.0c80-0d: {} `,
 	})
 	expectResult(t, "sel.StreamExecute", result, defaultSelectResult)
diff --git a/go/vt/vtgate/evalengine/convert_test.go b/go/vt/vtgate/evalengine/convert_test.go
index 2c8abba644..cf69778dbe 100644
--- a/go/vt/vtgate/evalengine/convert_test.go
+++ b/go/vt/vtgate/evalengine/convert_test.go
@@ -65,26 +65,26 @@ func TestConvertSimplification(t *testing.T) {
 		{"42", ok("INT64(42)"), ok("INT64(42)")},
 		{"1 + (1 + 1) * 8", ok("INT64(1) + ((INT64(1) + INT64(1)) * INT64(8))"), ok("INT64(17)")},
 		{"1.0e0 + (1 + 1) * 8.0e0", ok("FLOAT64(1) + ((INT64(1) + INT64(1)) * FLOAT64(8))"), ok("FLOAT64(17)")},
-		{"'pokemon' LIKE 'poke%'", ok("VARBINARY(\"pokemon\") LIKE VARBINARY(\"poke%\")"), ok("INT64(1)")},
+		{"'pokemon' LIKE 'poke%'", ok("VARCHAR(\"pokemon\") LIKE VARCHAR(\"poke%\")"), ok("INT64(1)")},
 		{
 			"'foo' COLLATE utf8mb4_general_ci IN ('bar' COLLATE latin1_swedish_ci, 'baz')",
-			ok(`VARBINARY("foo") COLLATE utf8mb4_general_ci IN (VARBINARY("bar") COLLATE latin1_swedish_ci, VARBINARY("baz"))`),
+			ok(`VARCHAR("foo") COLLATE utf8mb4_general_ci IN (VARCHAR("bar") COLLATE latin1_swedish_ci, VARCHAR("baz"))`),
 			err("COLLATION 'latin1_swedish_ci' is not valid for CHARACTER SET 'utf8mb4'"),
 		},
 		{`"pokemon" in ("bulbasaur", "venusaur", "charizard")`,
-			ok(`VARBINARY("pokemon") IN (VARBINARY("bulbasaur"), VARBINARY("venusaur"), VARBINARY("charizard"))`),
+			ok(`VARCHAR("pokemon") IN (VARCHAR("bulbasaur"), VARCHAR("venusaur"), VARCHAR("charizard"))`),
 			ok("INT64(0)"),
 		},
 		{`"pokemon" in ("bulbasaur", "venusaur", "pokemon")`,
-			ok(`VARBINARY("pokemon") IN (VARBINARY("bulbasaur"), VARBINARY("venusaur"), VARBINARY("pokemon"))`),
+			ok(`VARCHAR("pokemon") IN (VARCHAR("bulbasaur"), VARCHAR("venusaur"), VARCHAR("pokemon"))`),
 			ok("INT64(1)"),
 		},
 		{`"pokemon" in ("bulbasaur", "venusaur", "pokemon", NULL)`,
-			ok(`VARBINARY("pokemon") IN (VARBINARY("bulbasaur"), VARBINARY("venusaur"), VARBINARY("pokemon"), NULL)`),
+			ok(`VARCHAR("pokemon") IN (VARCHAR("bulbasaur"), VARCHAR("venusaur"), VARCHAR("pokemon"), NULL)`),
 			ok(`INT64(1)`),
 		},
 		{`"pokemon" in ("bulbasaur", "venusaur", NULL)`,
-			ok(`VARBINARY("pokemon") IN (VARBINARY("bulbasaur"), VARBINARY("venusaur"), NULL)`),
+			ok(`VARCHAR("pokemon") IN (VARCHAR("bulbasaur"), VARCHAR("venusaur"), NULL)`),
 			ok(`NULL`),
 		},
 		{"0 + NULL", ok("INT64(0) + NULL"), ok("NULL")},
@@ -93,6 +93,10 @@ func TestConvertSimplification(t *testing.T) {
 		{"1 + 0.05e0", ok("INT64(1) + FLOAT64(0.05)"), ok("FLOAT64(1.05)")},
 		{"1 / 1", ok("INT64(1) / INT64(1)"), ok("DECIMAL(1.0000)")},
 		{"(14620 / 9432456) / (24250 / 9432456)", ok("(INT64(14620) / INT64(9432456)) / (INT64(24250) / INT64(9432456))"), ok("DECIMAL(0.60288653)")},
+		{"COALESCE(NULL, 2, NULL, 4)", ok("COALESCE(NULL, INT64(2), NULL, INT64(4))"), ok("INT64(2)")},
+		{"coalesce(NULL, 2, NULL, 4)", ok("COALESCE(NULL, INT64(2), NULL, INT64(4))"), ok("INT64(2)")},
+		{"coalesce(NULL, NULL)", ok("COALESCE(NULL, NULL)"), ok("NULL")},
+		{"coalesce(NULL)", ok("COALESCE(NULL)"), ok("NULL")},
 	}
 
 	for _, tc := range testCases {
@@ -168,7 +172,7 @@ func TestEvaluate(t *testing.T) {
 		expected:   sqltypes.NewUint64(22),
 	}, {
 		expression: ":string_bind_variable",
-		expected:   sqltypes.NewVarBinary("bar"),
+		expected:   sqltypes.NewVarChar("bar"),
 	}, {
 		expression: ":float_bind_variable",
 		expected:   sqltypes.NewFloat64(2.2),
@@ -302,10 +306,10 @@ func TestEvaluateTuple(t *testing.T) {
 		expected:   []sqltypes.Value{sqltypes.NewInt64(1), sqltypes.NewInt64(2), sqltypes.NewInt64(4)},
 	}, {
 		expression: "(1,'2',4)",
-		expected:   []sqltypes.Value{sqltypes.NewInt64(1), sqltypes.NewVarBinary("2"), sqltypes.NewInt64(4)},
+		expected:   []sqltypes.Value{sqltypes.NewInt64(1), sqltypes.NewVarChar("2"), sqltypes.NewInt64(4)},
 	}, {
 		expression: "(1,'2',4.0)",
-		expected:   []sqltypes.Value{sqltypes.NewInt64(1), sqltypes.NewVarBinary("2"), sqltypes.NewDecimal("4.0")},
+		expected:   []sqltypes.Value{sqltypes.NewInt64(1), sqltypes.NewVarChar("2"), sqltypes.NewDecimal("4.0")},
 	}}
 
 	for _, test := range tests {
diff --git a/go/vt/vtgate/evalengine/integration/comparison_test.go b/go/vt/vtgate/evalengine/integration/comparison_test.go
index f20203b371..de3277a7f5 100644
--- a/go/vt/vtgate/evalengine/integration/comparison_test.go
+++ b/go/vt/vtgate/evalengine/integration/comparison_test.go
@@ -18,6 +18,8 @@ package integration
 
 import (
 	"fmt"
+	"math"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -44,19 +46,18 @@ func perm1(a []string, f func([]string), i int) {
 func compareRemoteQuery(t *testing.T, conn *mysql.Conn, query string) {
 	t.Helper()
 
-	local, _, localErr := safeEvaluate(query)
-	if localErr != nil {
-		t.Errorf("local failure: %v", localErr)
-		return
-	}
+	local, evaluated, localErr := safeEvaluate(query)
 	remote, remoteErr := conn.ExecuteFetch(query, 1, false)
-	if remoteErr != nil {
-		t.Errorf("remote failure: %v", remoteErr)
-		return
-	}
 
-	if local.Value().String() != remote.Rows[0][0].String() {
-		t.Errorf("mismatch for query %q: local=%v, remote=%v", query, local.Value().String(), remote.Rows[0][0].String())
+	var localVal, remoteVal string
+	if evaluated {
+		localVal = local.Value().String()
+	}
+	if remoteErr == nil {
+		remoteVal = remote.Rows[0][0].String()
+	}
+	if diff := compareResult(localErr, remoteErr, localVal, remoteVal, evaluated); diff != "" {
+		t.Errorf("%s\nquery: %s", diff, query)
 	}
 }
 
@@ -110,3 +111,108 @@ func TestAllIsStatements(t *testing.T) {
 		}
 	}
 }
+
+func genSubsets1(args []string, subset []string, a, b int, yield func([]string)) {
+	if a == len(subset) {
+		yield(subset)
+		return
+	}
+	if b >= len(args) {
+		return
+	}
+	subset[a] = args[b]
+	genSubsets1(args, subset, a+1, b+1, yield)
+	genSubsets1(args, subset, a+0, b+1, yield)
+}
+
+func genSubsets(args []string, subsetLen int, yield func([]string)) {
+	subset := make([]string, subsetLen)
+	genSubsets1(args, subset, 0, 0, yield)
+}
+
+func TestMultiComparisons(t *testing.T) {
+	var numbers = []string{
+		`0`, `-1`, `1`, `0.0`, `1.0`, `-1.0`, `1.0E0`, `-1.0E0`, `0.0E0`,
+		strconv.FormatUint(math.MaxUint64, 10),
+		strconv.FormatUint(math.MaxInt64, 10),
+		strconv.FormatInt(math.MinInt64, 10),
+		`'foobar'`, `'FOOBAR'`,
+		`"0"`, `"-1"`, `"1"`,
+		`_utf8mb4 'foobar'`, `_utf8mb4 'FOOBAR'`,
+		`_binary '0'`, `_binary '-1'`, `_binary '1'`,
+	}
+
+	var conn = mysqlconn(t)
+	defer conn.Close()
+
+	for _, method := range []string{"LEAST", "GREATEST"} {
+		for _, argc := range []int{2, 3} {
+			t.Run(fmt.Sprintf("%s(#%d)", method, argc), func(t *testing.T) {
+				genSubsets(numbers, argc, func(num []string) {
+					query := fmt.Sprintf("SELECT %s(%s)", method, strings.Join(num, ","))
+					compareRemoteQuery(t, conn, query)
+				})
+			})
+		}
+	}
+}
+
+func TestLikeComparison(t *testing.T) {
+	var left = []string{
+		`'foobar'`, `'FOOBAR'`,
+		`'1234'`, `1234`,
+		`_utf8mb4 'foobar' COLLATE utf8mb4_0900_as_cs`,
+		`_utf8mb4 'FOOBAR' COLLATE utf8mb4_0900_as_cs`,
+	}
+	var right = append([]string{
+		`'foo%'`, `'FOO%'`, `'foo_ar'`, `'FOO_AR'`,
+		`'12%'`, `'12_4'`,
+		`_utf8mb4 'foo%' COLLATE utf8mb4_0900_as_cs`,
+		`_utf8mb4 'FOO%' COLLATE utf8mb4_0900_as_cs`,
+		`_utf8mb4 'foo_ar' COLLATE utf8mb4_0900_as_cs`,
+		`_utf8mb4 'FOO_AR' COLLATE utf8mb4_0900_as_cs`,
+	}, left...)
+
+	var conn = mysqlconn(t)
+	defer conn.Close()
+
+	for _, lhs := range left {
+		for _, rhs := range right {
+			query := fmt.Sprintf("SELECT %s LIKE %s", lhs, rhs)
+			compareRemoteQuery(t, conn, query)
+		}
+	}
+}
+
+func TestTypes(t *testing.T) {
+	var conn = mysqlconn(t)
+	defer conn.Close()
+
+	var queries = []string{
+		"1 > 3",
+		"3 > 1",
+		"-1 > -1",
+		"1 = 1",
+		"-1 = 1",
+		"1 IN (1, -2, 3)",
+		"1 LIKE 1",
+		"-1 LIKE -1",
+		"-1 LIKE 1",
+		`"foo" IN ("bar", "FOO", "baz")`,
+		`'pokemon' LIKE 'poke%'`,
+		`(1, 2) = (1, 2)`,
+		`1 = 'sad'`,
+		`(1, 2) = (1, 3)`,
+		`LEAST(0,0.0)`,
+		`LEAST(0,1,0.0)`,
+		`LEAST(0.0,0)`,
+		`LEAST(0, 8446744073709551614)`,
+		`"foobar"`,
+		`X'444444'`,
+		`_binary "foobar"`,
+	}
+
+	for _, query := range queries {
+		compareRemoteQuery(t, conn, "SELECT "+query)
+	}
+}
diff --git a/go/vt/vtgate/evalengine/integration/fuzz_test.go b/go/vt/vtgate/evalengine/integration/fuzz_test.go
index 7d8c153ac4..8e25bb4f94 100644
--- a/go/vt/vtgate/evalengine/integration/fuzz_test.go
+++ b/go/vt/vtgate/evalengine/integration/fuzz_test.go
@@ -87,43 +87,6 @@ func (d dummyCollation) CollationIDLookup(_ sqlparser.Expr) collations.ID {
 	return collations.ID(d)
 }
 
-func TestTypes(t *testing.T) {
-	var conn = mysqlconn(t)
-	defer conn.Close()
-
-	var queries = []string{
-		"1 > 3",
-		"3 > 1",
-		"-1 > -1",
-		"1 = 1",
-		"-1 = 1",
-		"1 IN (1, -2, 3)",
-		"1 LIKE 1",
-		"-1 LIKE -1",
-		"-1 LIKE 1",
-		`"foo" IN ("bar", "FOO", "baz")`,
-		`'pokemon' LIKE 'poke%'`,
-		`(1, 2) = (1, 2)`,
-		`1 = 'sad'`,
-		`(1, 2) = (1, 3)`,
-	}
-
-	for _, query := range queries {
-		query = "SELECT " + query
-		remote, err := conn.ExecuteFetch(query, 1, false)
-		if err != nil {
-			t.Fatal(err)
-		}
-		local, _, err := safeEvaluate(query)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if local.Value().String() != remote.Rows[0][0].String() {
-			t.Errorf("mismatch for query %q: local=%v, remote=%v", query, local.Value().String(), remote.Rows[0][0].String())
-		}
-	}
-}
-
 var fuzzMaxTime = flag.Duration("fuzz-duration", 30*time.Second, "maximum time to fuzz for")
 var fuzzMaxFailures = flag.Int("fuzz-total", 0, "maximum number of failures to fuzz for")
 var fuzzSeed = flag.Int64("fuzz-seed", time.Now().Unix(), "RNG seed when generating fuzz expressions")
@@ -167,7 +130,7 @@ func safeEvaluate(query string) (evalengine.EvalResult, bool, error) {
 				err = fmt.Errorf("PANIC: %v", r)
 			}
 		}()
-		expr, err = evalengine.ConvertEx(astExpr, dummyCollation(45), true)
+		expr, err = evalengine.ConvertEx(astExpr, dummyCollation(8), true)
 		return
 	}()
 
@@ -321,29 +284,33 @@ type mismatch struct {
 	evaluated           bool
 }
 
-func (cr *mismatch) Error() string {
-	if cr.localErr != nil {
-		if cr.remoteErr == nil {
-			return fmt.Sprintf("%v (eval=%v); mysql response: %s", cr.localErr, cr.evaluated, cr.remoteVal)
+func compareResult(localErr, remoteErr error, localVal, remoteVal string, evaluated bool) string {
+	if localErr != nil {
+		if remoteErr == nil {
+			return fmt.Sprintf("%v (eval=%v); mysql response: %s", localErr, evaluated, remoteVal)
 		}
-		if !errorsMatch(cr.remoteErr, cr.localErr) {
-			return fmt.Sprintf("mismatch in errors: eval=%s; mysql response: %s", cr.localErr.Error(), cr.remoteErr.Error())
+		if !errorsMatch(remoteErr, localErr) {
+			return fmt.Sprintf("mismatch in errors: eval=%s; mysql response: %s", localErr.Error(), remoteErr.Error())
 		}
 		return ""
 	}
 
-	if cr.remoteErr != nil {
+	if remoteErr != nil {
 		for _, ke := range knownErrors {
-			if ke.MatchString(cr.remoteErr.Error()) {
+			if ke.MatchString(remoteErr.Error()) {
 				return ""
 			}
 		}
-		return fmt.Sprintf("%v (eval=%v); mysql failed with: %s", cr.localVal, cr.evaluated, cr.remoteErr.Error())
+		return fmt.Sprintf("%v (eval=%v); mysql failed with: %s", localVal, evaluated, remoteErr.Error())
 	}
 
-	if cr.localVal != cr.remoteVal {
-		return fmt.Sprintf("different results:%s (eval=%v); mysql response: %s", cr.localVal, cr.evaluated, cr.remoteVal)
+	if localVal != remoteVal {
+		return fmt.Sprintf("different results:%s (eval=%v); mysql response: %s", localVal, evaluated, remoteVal)
 	}
 
 	return ""
 }
+
+func (cr *mismatch) Error() string {
+	return compareResult(cr.localErr, cr.remoteErr, cr.localVal, cr.remoteVal, cr.evaluated)
+}
diff --git a/go/vt/vtgate/evalengine/mysql_test.go b/go/vt/vtgate/evalengine/mysql_test.go
index accc62d660..f63f8414cd 100644
--- a/go/vt/vtgate/evalengine/mysql_test.go
+++ b/go/vt/vtgate/evalengine/mysql_test.go
@@ -118,24 +118,6 @@ func TestMySQLGolden(t *testing.T) {
 
 func TestDebug1(t *testing.T) {
 	// Debug
-	eval, err := testSingle(t, `SELECT ((NULL, ("foo" >= -1), "FOO", NULL), "fOo", ("foo" < ("fOo", -1, "FOO", 0)), NULL) <=> (0, "foo", 0, NULL)`)
-	t.Logf("eval=%s err=%v", eval.Value(), err) // want value=""
-}
-
-func TestDebug2(t *testing.T) {
-	// Debug
-	eval, err := testSingle(t, `SELECT ((("FOO" / "FOO") + -1) + (("foo", ((NULL, -1, 1, 1) * -1), ("foo", 0, NULL, NULL), "fOo") >= "foo")) + "fOo"`)
-	t.Logf("eval=%s err=%v", eval.Value(), err) // want value=""
-}
-
-func TestDebug3(t *testing.T) {
-	// Debug
-	eval, err := testSingle(t, `SELECT (0, ("fOo" NOT LIKE ((NULL LIKE "foo"), "foo", -1, NULL)), "foo", "fOo") <=> "fOo"`)
-	t.Logf("eval=%s err=%v", eval.Value(), err) // want value=""
-}
-
-func TestDebug4(t *testing.T) {
-	// Debug
-	eval, err := testSingle(t, `SELECT ("foo", (("foo" >= "fOo"), 0, ((1, (NULL >= "foo"), (1 LIKE "fOo"), "fOo"), "fOo", (-1 <=> "FOO"), 1), NULL), 0, 1) != ("FOO", ("FOO", 1, NULL, -1), 1, (NULL != NULL))`)
+	eval, err := testSingle(t, `SELECT LEAST(0.0,'foobar')`)
 	t.Logf("eval=%s err=%v", eval.Value(), err) // want value=""
 }
diff --git a/go/vt/vtgate/executor_dml_test.go b/go/vt/vtgate/executor_dml_test.go
index 2415b7e91c..3df9448bd6 100644
--- a/go/vt/vtgate/executor_dml_test.go
+++ b/go/vt/vtgate/executor_dml_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package vtgate
 
 import (
+	"context"
 	"reflect"
 	"strings"
 	"testing"
@@ -30,8 +31,6 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"context"
-
 	"vitess.io/vitess/go/sqltypes"
 	_ "vitess.io/vitess/go/vt/vtgate/vindexes"
 	"vitess.io/vitess/go/vt/vttablet/sandboxconn"
@@ -115,7 +114,7 @@ func TestUpdateEqual(t *testing.T) {
 		{
 			Sql: "delete from name_lastname_keyspace_id_map where `name` = :name and lastname = :lastname and keyspace_id = :keyspace_id",
 			BindVariables: map[string]*querypb.BindVariable{
-				"lastname":    sqltypes.ValueBindVariable(sqltypes.NewVarChar("foo")),
+				"lastname":    sqltypes.StringBindVariable("foo"),
 				"name":        sqltypes.Int32BindVariable(1),
 				"keyspace_id": sqltypes.BytesBindVariable([]byte("\x16k@\xb4J\xbaK\xd6")),
 			},
@@ -123,8 +122,8 @@ func TestUpdateEqual(t *testing.T) {
 		{
 			Sql: "insert into name_lastname_keyspace_id_map(`name`, lastname, keyspace_id) values (:name_0, :lastname_0, :keyspace_id_0)",
 			BindVariables: map[string]*querypb.BindVariable{
-				"name_0":        sqltypes.BytesBindVariable([]byte("myname")),
-				"lastname_0":    sqltypes.BytesBindVariable([]byte("mylastname")),
+				"name_0":        sqltypes.StringBindVariable("myname"),
+				"lastname_0":    sqltypes.StringBindVariable("mylastname"),
 				"keyspace_id_0": sqltypes.BytesBindVariable([]byte("\x16k@\xb4J\xbaK\xd6")),
 			},
 		},
@@ -805,7 +804,7 @@ func TestInsertSharded(t *testing.T) {
 		Sql: "insert into `user`(id, v, `name`) values (:_Id_0, 2, :_name_0)",
 		BindVariables: map[string]*querypb.BindVariable{
 			"_Id_0":   sqltypes.Int64BindVariable(1),
-			"_name_0": sqltypes.BytesBindVariable([]byte("myname")),
+			"_name_0": sqltypes.StringBindVariable("myname"),
 			"__seq0":  sqltypes.Int64BindVariable(1),
 		},
 	}}
@@ -814,7 +813,7 @@ func TestInsertSharded(t *testing.T) {
 	wantQueries = []*querypb.BoundQuery{{
 		Sql: "insert into name_user_map(`name`, user_id) values (:name_0, :user_id_0)",
 		BindVariables: map[string]*querypb.BindVariable{
-			"name_0":    sqltypes.BytesBindVariable([]byte("myname")),
+			"name_0":    sqltypes.StringBindVariable("myname"),
 			"user_id_0": sqltypes.Uint64BindVariable(1),
 		},
 	}}
@@ -832,7 +831,7 @@ func TestInsertSharded(t *testing.T) {
 		BindVariables: map[string]*querypb.BindVariable{
 			"_Id_0":   sqltypes.Int64BindVariable(3),
 			"__seq0":  sqltypes.Int64BindVariable(3),
-			"_name_0": sqltypes.BytesBindVariable([]byte("myname2")),
+			"_name_0": sqltypes.StringBindVariable("myname2"),
 		},
 	}}
 	assertQueries(t, sbc2, wantQueries)
@@ -840,7 +839,7 @@ func TestInsertSharded(t *testing.T) {
 	wantQueries = []*querypb.BoundQuery{{
 		Sql: "insert into name_user_map(`name`, user_id) values (:name_0, :user_id_0)",
 		BindVariables: map[string]*querypb.BindVariable{
-			"name_0":    sqltypes.BytesBindVariable([]byte("myname2")),
+			"name_0":    sqltypes.StringBindVariable("myname2"),
 			"user_id_0": sqltypes.Uint64BindVariable(3),
 		},
 	}}
@@ -855,8 +854,8 @@ func TestInsertSharded(t *testing.T) {
 		Sql: "insert into user2(id, `name`, lastname) values (:_id_0, :_name_0, :_lastname_0)",
 		BindVariables: map[string]*querypb.BindVariable{
 			"_id_0":       sqltypes.Int64BindVariable(2),
-			"_name_0":     sqltypes.BytesBindVariable([]byte("myname")),
-			"_lastname_0": sqltypes.BytesBindVariable([]byte("mylastname")),
+			"_name_0":     sqltypes.StringBindVariable("myname"),
+			"_lastname_0": sqltypes.StringBindVariable("mylastname"),
 		},
 	}}
 	assertQueries(t, sbc1, wantQueries)
@@ -878,7 +877,7 @@ func TestInsertSharded(t *testing.T) {
 			"__seq0":  sqltypes.Int64BindVariable(1),
 			"vtg1":    sqltypes.Int64BindVariable(1),
 			"vtg2":    sqltypes.Int64BindVariable(2),
-			"vtg3":    sqltypes.BytesBindVariable([]byte("myname")),
+			"vtg3":    sqltypes.StringBindVariable("myname"),
 		},
 	}}
 	assertQueries(t, sbc1, wantQueries)
@@ -958,7 +957,7 @@ func TestInsertShardedAutocommitLookup(t *testing.T) {
 		Sql: "insert into `user`(id, v, `name`) values (:_Id_0, 2, :_name_0)",
 		BindVariables: map[string]*querypb.BindVariable{
 			"_Id_0":   sqltypes.Int64BindVariable(1),
-			"_name_0": sqltypes.BytesBindVariable([]byte("myname")),
+			"_name_0": sqltypes.StringBindVariable("myname"),
 			"__seq0":  sqltypes.Int64BindVariable(1),
 		},
 	}}
@@ -967,7 +966,7 @@ func TestInsertShardedAutocommitLookup(t *testing.T) {
 	wantQueries = []*querypb.BoundQuery{{
 		Sql: "insert into name_user_map(`name`, user_id) values (:name_0, :user_id_0) on duplicate key update `name` = values(`name`), user_id = values(user_id)",
 		BindVariables: map[string]*querypb.BindVariable{
-			"name_0":    sqltypes.BytesBindVariable([]byte("myname")),
+			"name_0":    sqltypes.StringBindVariable("myname"),
 			"user_id_0": sqltypes.Uint64BindVariable(1),
 		},
 	}}
@@ -1195,7 +1194,7 @@ func TestInsertComments(t *testing.T) {
 		Sql: "insert into `user`(id, v, `name`) values (:_Id_0, 2, :_name_0) /* trailing */",
 		BindVariables: map[string]*querypb.BindVariable{
 			"_Id_0":   sqltypes.Int64BindVariable(1),
-			"_name_0": sqltypes.BytesBindVariable([]byte("myname")),
+			"_name_0": sqltypes.StringBindVariable("myname"),
 			"__seq0":  sqltypes.Int64BindVariable(1),
 		},
 	}}
@@ -1204,7 +1203,7 @@ func TestInsertComments(t *testing.T) {
 	wantQueries = []*querypb.BoundQuery{{
 		Sql: "insert into name_user_map(`name`, user_id) values (:name_0, :user_id_0) /* trailing */",
 		BindVariables: map[string]*querypb.BindVariable{
-			"name_0":    sqltypes.BytesBindVariable([]byte("myname")),
+			"name_0":    sqltypes.StringBindVariable("myname"),
 			"user_id_0": sqltypes.Uint64BindVariable(1),
 		},
 	}}
@@ -1228,7 +1227,7 @@ func TestInsertGeneratorSharded(t *testing.T) {
 		BindVariables: map[string]*querypb.BindVariable{
 			"_Id_0":   sqltypes.Int64BindVariable(1),
 			"__seq0":  sqltypes.Int64BindVariable(1),
-			"_name_0": sqltypes.BytesBindVariable([]byte("myname")),
+			"_name_0": sqltypes.StringBindVariable("myname"),
 		},
 	}}
 	assertQueries(t, sbc, wantQueries)
@@ -1238,7 +1237,7 @@ func TestInsertGeneratorSharded(t *testing.T) {
 	}, {
 		Sql: "insert into name_user_map(`name`, user_id) values (:name_0, :user_id_0)",
 		BindVariables: map[string]*querypb.BindVariable{
-			"name_0":    sqltypes.BytesBindVariable([]byte("myname")),
+			"name_0":    sqltypes.StringBindVariable("myname"),
 			"user_id_0": sqltypes.Uint64BindVariable(1),
 		},
 	}}
@@ -1506,10 +1505,10 @@ func TestMultiInsertSharded(t *testing.T) {
 		Sql: "insert into `user`(id, v, `name`) values (:_Id_0, 1, :_name_0)",
 		BindVariables: map[string]*querypb.BindVariable{
 			"_Id_0":   sqltypes.Int64BindVariable(1),
-			"_name_0": sqltypes.BytesBindVariable([]byte("myname1")),
+			"_name_0": sqltypes.StringBindVariable("myname1"),
 			"__seq0":  sqltypes.Int64BindVariable(1),
 			"_Id_1":   sqltypes.Int64BindVariable(3),
-			"_name_1": sqltypes.BytesBindVariable([]byte("myname3")),
+			"_name_1": sqltypes.StringBindVariable("myname3"),
 			"__seq1":  sqltypes.Int64BindVariable(3),
 		},
 	}}
@@ -1518,10 +1517,10 @@ func TestMultiInsertSharded(t *testing.T) {
 		Sql: "insert into `user`(id, v, `name`) values (:_Id_1, 3, :_name_1)",
 		BindVariables: map[string]*querypb.BindVariable{
 			"_Id_0":   sqltypes.Int64BindVariable(1),
-			"_name_0": sqltypes.BytesBindVariable([]byte("myname1")),
+			"_name_0": sqltypes.StringBindVariable("myname1"),
 			"__seq0":  sqltypes.Int64BindVariable(1),
 			"_Id_1":   sqltypes.Int64BindVariable(3),
-			"_name_1": sqltypes.BytesBindVariable([]byte("myname3")),
+			"_name_1": sqltypes.StringBindVariable("myname3"),
 			"__seq1":  sqltypes.Int64BindVariable(3),
 		},
 	}}
@@ -1531,9 +1530,9 @@ func TestMultiInsertSharded(t *testing.T) {
 	wantQueries1 = []*querypb.BoundQuery{{
 		Sql: "insert into name_user_map(`name`, user_id) values (:name_0, :user_id_0), (:name_1, :user_id_1)",
 		BindVariables: map[string]*querypb.BindVariable{
-			"name_0":    sqltypes.BytesBindVariable([]byte("myname1")),
+			"name_0":    sqltypes.StringBindVariable("myname1"),
 			"user_id_0": sqltypes.Uint64BindVariable(1),
-			"name_1":    sqltypes.BytesBindVariable([]byte("myname3")),
+			"name_1":    sqltypes.StringBindVariable("myname3"),
 			"user_id_1": sqltypes.Uint64BindVariable(3),
 		},
 	}}
@@ -1549,10 +1548,10 @@ func TestMultiInsertSharded(t *testing.T) {
 		BindVariables: map[string]*querypb.BindVariable{
 			"_Id_0":   sqltypes.Int64BindVariable(1),
 			"__seq0":  sqltypes.Int64BindVariable(1),
-			"_name_0": sqltypes.BytesBindVariable([]byte("myname1")),
+			"_name_0": sqltypes.StringBindVariable("myname1"),
 			"_Id_1":   sqltypes.Int64BindVariable(2),
 			"__seq1":  sqltypes.Int64BindVariable(2),
-			"_name_1": sqltypes.BytesBindVariable([]byte("myname2")),
+			"_name_1": sqltypes.StringBindVariable("myname2"),
 		},
 	}}
 
@@ -1561,9 +1560,9 @@ func TestMultiInsertSharded(t *testing.T) {
 	wantQueries = []*querypb.BoundQuery{{
 		Sql: "insert into name_user_map(`name`, user_id) values (:name_0, :user_id_0), (:name_1, :user_id_1)",
 		BindVariables: map[string]*querypb.BindVariable{
-			"name_0":    sqltypes.BytesBindVariable([]byte("myname1")),
+			"name_0":    sqltypes.StringBindVariable("myname1"),
 			"user_id_0": sqltypes.Uint64BindVariable(1),
-			"name_1":    sqltypes.BytesBindVariable([]byte("myname2")),
+			"name_1":    sqltypes.StringBindVariable("myname2"),
 			"user_id_1": sqltypes.Uint64BindVariable(2),
 		},
 	}}
@@ -1579,22 +1578,22 @@ func TestMultiInsertSharded(t *testing.T) {
 		Sql: "insert into user2(id, `name`, lastname) values (:_id_0, :_name_0, :_lastname_0)",
 		BindVariables: map[string]*querypb.BindVariable{
 			"_id_0":       sqltypes.Int64BindVariable(2),
-			"_name_0":     sqltypes.BytesBindVariable([]byte("myname")),
-			"_lastname_0": sqltypes.BytesBindVariable([]byte("mylastname")),
+			"_name_0":     sqltypes.StringBindVariable("myname"),
+			"_lastname_0": sqltypes.StringBindVariable("mylastname"),
 			"_id_1":       sqltypes.Int64BindVariable(3),
-			"_name_1":     sqltypes.BytesBindVariable([]byte("myname2")),
-			"_lastname_1": sqltypes.BytesBindVariable([]byte("mylastname2")),
+			"_name_1":     sqltypes.StringBindVariable("myname2"),
+			"_lastname_1": sqltypes.StringBindVariable("mylastname2"),
 		},
 	}}
 	assertQueries(t, sbc1, wantQueries)
 	wantQueries = []*querypb.BoundQuery{{
 		Sql: "insert into name_lastname_keyspace_id_map(`name`, lastname, keyspace_id) values (:name_0, :lastname_0, :keyspace_id_0), (:name_1, :lastname_1, :keyspace_id_1)",
 		BindVariables: map[string]*querypb.BindVariable{
-			"name_0":        sqltypes.BytesBindVariable([]byte("myname")),
-			"lastname_0":    sqltypes.BytesBindVariable([]byte("mylastname")),
+			"name_0":        sqltypes.StringBindVariable("myname"),
+			"lastname_0":    sqltypes.StringBindVariable("mylastname"),
 			"keyspace_id_0": sqltypes.BytesBindVariable([]byte("\x06\xe7\xea\"Βp\x8f")),
-			"name_1":        sqltypes.BytesBindVariable([]byte("myname2")),
-			"lastname_1":    sqltypes.BytesBindVariable([]byte("mylastname2")),
+			"name_1":        sqltypes.StringBindVariable("myname2"),
+			"lastname_1":    sqltypes.StringBindVariable("mylastname2"),
 			"keyspace_id_1": sqltypes.BytesBindVariable([]byte("N\xb1\x90ɢ\xfa\x16\x9c")),
 		},
 	}}
diff --git a/go/vt/vtgate/executor_select_test.go b/go/vt/vtgate/executor_select_test.go
index fe451b2eb2..af0284f568 100644
--- a/go/vt/vtgate/executor_select_test.go
+++ b/go/vt/vtgate/executor_select_test.go
@@ -432,13 +432,13 @@ func TestSelectSystemVariables(t *testing.T) {
 			{Name: "@@skip_query_plan_cache", Type: sqltypes.Int64},
 			{Name: "@@enable_system_settings", Type: sqltypes.Int64},
 			{Name: "@@sql_select_limit", Type: sqltypes.Int64},
-			{Name: "@@transaction_mode", Type: sqltypes.VarBinary},
-			{Name: "@@workload", Type: sqltypes.VarBinary},
-			{Name: "@@read_after_write_gtid", Type: sqltypes.VarBinary},
+			{Name: "@@transaction_mode", Type: sqltypes.VarChar},
+			{Name: "@@workload", Type: sqltypes.VarChar},
+			{Name: "@@read_after_write_gtid", Type: sqltypes.VarChar},
 			{Name: "@@read_after_write_timeout", Type: sqltypes.Float64},
-			{Name: "@@session_track_gtids", Type: sqltypes.VarBinary},
-			{Name: "@@ddl_strategy", Type: sqltypes.VarBinary},
-			{Name: "@@socket", Type: sqltypes.VarBinary},
+			{Name: "@@session_track_gtids", Type: sqltypes.VarChar},
+			{Name: "@@ddl_strategy", Type: sqltypes.VarChar},
+			{Name: "@@socket", Type: sqltypes.VarChar},
 		},
 		Rows: [][]sqltypes.Value{{
 			// the following are the uninitialised session values
@@ -447,14 +447,14 @@ func TestSelectSystemVariables(t *testing.T) {
 			sqltypes.NewInt64(0),
 			sqltypes.NewInt64(0),
 			sqltypes.NewInt64(0),
-			sqltypes.NewVarBinary("UNSPECIFIED"),
-			sqltypes.NewVarBinary(""),
+			sqltypes.NewVarChar("UNSPECIFIED"),
+			sqltypes.NewVarChar(""),
 			// these have been set at the beginning of the test
-			sqltypes.NewVarBinary("a fine gtid"),
+			sqltypes.NewVarChar("a fine gtid"),
 			sqltypes.NewFloat64(13),
-			sqltypes.NewVarBinary("own_gtid"),
-			sqltypes.NewVarBinary(""),
-			sqltypes.NewVarBinary(""),
+			sqltypes.NewVarChar("own_gtid"),
+			sqltypes.NewVarChar(""),
+			sqltypes.NewVarChar(""),
 		}},
 	}
 	require.NoError(t, err)
@@ -516,10 +516,10 @@ func TestSelectUserDefindVariable(t *testing.T) {
 	require.NoError(t, err)
 	wantResult = &sqltypes.Result{
 		Fields: []*querypb.Field{
-			{Name: "@foo", Type: sqltypes.VarBinary},
+			{Name: "@foo", Type: sqltypes.VarChar},
 		},
 		Rows: [][]sqltypes.Value{{
-			sqltypes.NewVarBinary("bar"),
+			sqltypes.NewVarChar("bar"),
 		}},
 	}
 	utils.MustMatch(t, wantResult, result, "Mismatch")
@@ -670,10 +670,10 @@ func TestSelectDatabase(t *testing.T) {
 		map[string]*querypb.BindVariable{})
 	wantResult := &sqltypes.Result{
 		Fields: []*querypb.Field{
-			{Name: "database()", Type: sqltypes.VarBinary},
+			{Name: "database()", Type: sqltypes.VarChar},
 		},
 		Rows: [][]sqltypes.Value{{
-			sqltypes.NewVarBinary("TestExecutor@primary"),
+			sqltypes.NewVarChar("TestExecutor@primary"),
 		}},
 	}
 	require.NoError(t, err)
@@ -711,15 +711,15 @@ func TestSelectBindvars(t *testing.T) {
 	// Test with StringBindVariable
 	sql = "select id from user where name in (:name1, :name2)"
 	_, err = executorExec(executor, sql, map[string]*querypb.BindVariable{
-		"name1": sqltypes.BytesBindVariable([]byte("foo1")),
-		"name2": sqltypes.BytesBindVariable([]byte("foo2")),
+		"name1": sqltypes.StringBindVariable("foo1"),
+		"name2": sqltypes.StringBindVariable("foo2"),
 	})
 	require.NoError(t, err)
 	wantQueries = []*querypb.BoundQuery{{
 		Sql: "select id from `user` where `name` in ::__vals",
 		BindVariables: map[string]*querypb.BindVariable{
-			"name1":  sqltypes.BytesBindVariable([]byte("foo1")),
-			"name2":  sqltypes.BytesBindVariable([]byte("foo2")),
+			"name1":  sqltypes.StringBindVariable("foo1"),
+			"name2":  sqltypes.StringBindVariable("foo2"),
 			"__vals": sqltypes.TestBindVariable([]interface{}{"foo1", "foo2"}),
 		},
 	}}
@@ -775,7 +775,7 @@ func TestSelectBindvars(t *testing.T) {
 	}}
 	utils.MustMatch(t, wantQueries, sbc1.Queries)
 
-	vars, err := sqltypes.BuildBindVariable([]interface{}{sqltypes.NewVarBinary("nonexistent")})
+	vars, err := sqltypes.BuildBindVariable([]interface{}{sqltypes.NewVarChar("nonexistent")})
 	require.NoError(t, err)
 	wantLookupQueries := []*querypb.BoundQuery{{
 		Sql: "select `name`, user_id from name_user_map where `name` in ::name",
@@ -847,7 +847,7 @@ func TestSelectEqual(t *testing.T) {
 		BindVariables: map[string]*querypb.BindVariable{},
 	}}
 	utils.MustMatch(t, wantQueries, sbc1.Queries)
-	vars, err := sqltypes.BuildBindVariable([]interface{}{sqltypes.NewVarBinary("foo")})
+	vars, err := sqltypes.BuildBindVariable([]interface{}{sqltypes.NewVarChar("foo")})
 	require.NoError(t, err)
 	wantQueries = []*querypb.BoundQuery{{
 		Sql: "select `name`, user_id from name_user_map where `name` in ::name",
@@ -1061,7 +1061,7 @@ func TestSelectIN(t *testing.T) {
 		BindVariables: map[string]*querypb.BindVariable{},
 	}}
 	utils.MustMatch(t, wantQueries, sbc1.Queries)
-	vars, err := sqltypes.BuildBindVariable([]interface{}{sqltypes.NewVarBinary("foo")})
+	vars, err := sqltypes.BuildBindVariable([]interface{}{sqltypes.NewVarChar("foo")})
 	require.NoError(t, err)
 	wantQueries = []*querypb.BoundQuery{{
 		Sql: "select `name`, user_id from name_user_map where `name` in ::name",
@@ -1106,7 +1106,7 @@ func TestStreamSelectIN(t *testing.T) {
 		t.Errorf("result: %+v, want %+v", result, wantResult)
 	}
 
-	vars, err := sqltypes.BuildBindVariable([]interface{}{sqltypes.NewVarBinary("foo")})
+	vars, err := sqltypes.BuildBindVariable([]interface{}{sqltypes.NewVarChar("foo")})
 	require.NoError(t, err)
 	wantQueries := []*querypb.BoundQuery{{
 		Sql: "select `name`, user_id from name_user_map where `name` in ::name",
diff --git a/go/vt/vtgate/executor_set_test.go b/go/vt/vtgate/executor_set_test.go
index a26174cb6d..e3d5313621 100644
--- a/go/vt/vtgate/executor_set_test.go
+++ b/go/vt/vtgate/executor_set_test.go
@@ -163,7 +163,7 @@ func TestExecutorSet(t *testing.T) {
 		out: &vtgatepb.Session{Autocommit: true, Options: &querypb.ExecuteOptions{SqlSelectLimit: 0}},
 	}, {
 		in:  "set sql_select_limit = 'asdfasfd'",
-		err: "incorrect argument type to variable 'sql_select_limit': VARBINARY",
+		err: "incorrect argument type to variable 'sql_select_limit': VARCHAR",
 	}, {
 		in:  "set autocommit = 1+1",
 		err: "variable 'autocommit' can't be set to the value: 2 is not a boolean",
@@ -455,7 +455,7 @@ func TestPlanExecutorSetUDV(t *testing.T) {
 func TestSetUDVFromTabletInput(t *testing.T) {
 	executor, sbc1, _, _ := createLegacyExecutorEnv()
 
-	fields := sqltypes.MakeTestFields("some", "VARBINARY")
+	fields := sqltypes.MakeTestFields("some", "VARCHAR")
 	sbc1.SetResults([]*sqltypes.Result{
 		sqltypes.MakeTestResult(
 			fields,
@@ -470,7 +470,7 @@ func TestSetUDVFromTabletInput(t *testing.T) {
 	_, err := executorExec(executor, "set @foo = concat('a','b','c')", nil)
 	require.NoError(t, err)
 
-	want := map[string]*querypb.BindVariable{"foo": sqltypes.StringBindVariable("abc")}
+	want := map[string]*querypb.BindVariable{"foo": sqltypes.BytesBindVariable([]byte("abc"))}
 	utils.MustMatch(t, want, primarySession.UserDefinedVariables, "")
 }
 
diff --git a/go/vt/vtgate/logstats_test.go b/go/vt/vtgate/logstats_test.go
index 0bb1443e1a..84c37384a7 100644
--- a/go/vt/vtgate/logstats_test.go
+++ b/go/vt/vtgate/logstats_test.go
@@ -18,6 +18,7 @@ package vtgate
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"errors"
 	"net/url"
@@ -25,8 +26,6 @@ import (
 	"testing"
 	"time"
 
-	"context"
-
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/streamlog"
 	"vitess.io/vitess/go/vt/callinfo"
@@ -106,7 +105,7 @@ func TestLogStatsFormat(t *testing.T) {
 
 	*streamlog.QueryLogFormat = "text"
 	got = testFormat(logStats, url.Values(params))
-	want = "test\t\t\t''\t''\t2017-01-01 01:02:03.000000\t2017-01-01 01:02:04.000001\t1.000001\t0.000000\t0.000000\t0.000000\t\t\"sql1\"\tmap[strVal:type:VARBINARY value:\"abc\"]\t0\t0\t\"\"\t\"ks\"\t\"table\"\t\"PRIMARY\"\t\n"
+	want = "test\t\t\t''\t''\t2017-01-01 01:02:03.000000\t2017-01-01 01:02:04.000001\t1.000001\t0.000000\t0.000000\t0.000000\t\t\"sql1\"\tmap[strVal:type:VARCHAR value:\"abc\"]\t0\t0\t\"\"\t\"ks\"\t\"table\"\t\"PRIMARY\"\t\n"
 	if got != want {
 		t.Errorf("logstats format: got:\n%q\nwant:\n%q\n", got, want)
 	}
@@ -121,7 +120,7 @@ func TestLogStatsFormat(t *testing.T) {
 	if err != nil {
 		t.Errorf("logstats format: error marshaling json: %v -- got:\n%v", err, got)
 	}
-	want = "{\n    \"BindVars\": {\n        \"strVal\": {\n            \"type\": \"VARBINARY\",\n            \"value\": \"abc\"\n        }\n    },\n    \"CommitTime\": 0,\n    \"Effective Caller\": \"\",\n    \"End\": \"2017-01-01 01:02:04.000001\",\n    \"Error\": \"\",\n    \"ExecuteTime\": 0,\n    \"ImmediateCaller\": \"\",\n    \"Keyspace\": \"ks\",\n    \"Method\": \"test\",\n    \"PlanTime\": 0,\n    \"RemoteAddr\": \"\",\n    \"RowsAffected\": 0,\n    \"SQL\": \"sql1\",\n    \"ShardQueries\": 0,\n    \"Start\": \"2017-01-01 01:02:03.000000\",\n    \"StmtType\": \"\",\n    \"Table\": \"table\",\n    \"TabletType\": \"PRIMARY\",\n    \"TotalTime\": 1.000001,\n    \"Username\": \"\"\n}"
+	want = "{\n    \"BindVars\": {\n        \"strVal\": {\n            \"type\": \"VARCHAR\",\n            \"value\": \"abc\"\n        }\n    },\n    \"CommitTime\": 0,\n    \"Effective Caller\": \"\",\n    \"End\": \"2017-01-01 01:02:04.000001\",\n    \"Error\": \"\",\n    \"ExecuteTime\": 0,\n    \"ImmediateCaller\": \"\",\n    \"Keyspace\": \"ks\",\n    \"Method\": \"test\",\n    \"PlanTime\": 0,\n    \"RemoteAddr\": \"\",\n    \"RowsAffected\": 0,\n    \"SQL\": \"sql1\",\n    \"ShardQueries\": 0,\n    \"Start\": \"2017-01-01 01:02:03.000000\",\n    \"StmtType\": \"\",\n    \"Table\": \"table\",\n    \"TabletType\": \"PRIMARY\",\n    \"TotalTime\": 1.000001,\n    \"Username\": \"\"\n}"
 	if string(formatted) != want {
 		t.Errorf("logstats format: got:\n%q\nwant:\n%v\n", string(formatted), want)
 	}
diff --git a/go/vt/vtgate/planbuilder/plan_test.go b/go/vt/vtgate/planbuilder/plan_test.go
index abe7f387ca..f5106c00b3 100644
--- a/go/vt/vtgate/planbuilder/plan_test.go
+++ b/go/vt/vtgate/planbuilder/plan_test.go
@@ -189,19 +189,28 @@ const (
 	gen4ErrorPrefix = "Gen4 error: "
 )
 
+func makeTestOutput(t *testing.T) string {
+	testOutputTempDir, err := os.MkdirTemp("testdata", "plan_test")
+	require.NoError(t, err)
+
+	t.Cleanup(func() {
+		if !t.Failed() {
+			_ = os.RemoveAll(testOutputTempDir)
+		} else {
+			t.Logf("Errors found in plantests. If the output is correct, run `cp %s/* testdata/` to update test expectations", testOutputTempDir)
+		}
+	})
+
+	return testOutputTempDir
+}
+
 func TestPlan(t *testing.T) {
 	vschemaWrapper := &vschemaWrapper{
 		v:             loadSchema(t, "schema_test.json", true),
 		sysVarEnabled: true,
 	}
+	testOutputTempDir := makeTestOutput(t)
 
-	testOutputTempDir, err := os.MkdirTemp("", "plan_test")
-	require.NoError(t, err)
-	defer func() {
-		if !t.Failed() {
-			os.RemoveAll(testOutputTempDir)
-		}
-	}()
 	// You will notice that some tests expect user.Id instead of user.id.
 	// This is because we now pre-create vindex columns in the symbol
 	// table, which come from vschema. In the test vschema,
@@ -238,10 +247,7 @@ func TestSysVarSetDisabled(t *testing.T) {
 		sysVarEnabled: false,
 	}
 
-	testOutputTempDir, err := os.MkdirTemp("", "plan_test")
-	require.NoError(t, err)
-	defer os.RemoveAll(testOutputTempDir)
-	testFile(t, "set_sysvar_disabled_cases.txt", testOutputTempDir, vschemaWrapper)
+	testFile(t, "set_sysvar_disabled_cases.txt", makeTestOutput(t), vschemaWrapper)
 }
 
 func TestOne(t *testing.T) {
@@ -258,15 +264,7 @@ func TestRubyOnRailsQueries(t *testing.T) {
 		sysVarEnabled: true,
 	}
 
-	testOutputTempDir, err := os.MkdirTemp("", "plan_test")
-	require.NoError(t, err)
-	defer func() {
-		if !t.Failed() {
-			os.RemoveAll(testOutputTempDir)
-		}
-	}()
-
-	testFile(t, "rails_cases.txt", testOutputTempDir, vschemaWrapper)
+	testFile(t, "rails_cases.txt", makeTestOutput(t), vschemaWrapper)
 }
 
 func TestOLTP(t *testing.T) {
@@ -275,15 +273,7 @@ func TestOLTP(t *testing.T) {
 		sysVarEnabled: true,
 	}
 
-	testOutputTempDir, err := os.MkdirTemp("", "plan_test")
-	require.NoError(t, err)
-	defer func() {
-		if !t.Failed() {
-			os.RemoveAll(testOutputTempDir)
-		}
-	}()
-
-	testFile(t, "oltp_cases.txt", testOutputTempDir, vschemaWrapper)
+	testFile(t, "oltp_cases.txt", makeTestOutput(t), vschemaWrapper)
 }
 
 func TestTPCC(t *testing.T) {
@@ -292,15 +282,7 @@ func TestTPCC(t *testing.T) {
 		sysVarEnabled: true,
 	}
 
-	testOutputTempDir, err := os.MkdirTemp("", "plan_test")
-	require.NoError(t, err)
-	defer func() {
-		if !t.Failed() {
-			os.RemoveAll(testOutputTempDir)
-		}
-	}()
-
-	testFile(t, "tpcc_cases.txt", testOutputTempDir, vschemaWrapper)
+	testFile(t, "tpcc_cases.txt", makeTestOutput(t), vschemaWrapper)
 }
 
 func TestTPCH(t *testing.T) {
@@ -309,15 +291,7 @@ func TestTPCH(t *testing.T) {
 		sysVarEnabled: true,
 	}
 
-	testOutputTempDir, err := os.MkdirTemp("", "plan_test")
-	require.NoError(t, err)
-	defer func() {
-		if !t.Failed() {
-			os.RemoveAll(testOutputTempDir)
-		}
-	}()
-
-	testFile(t, "tpch_cases.txt", testOutputTempDir, vschemaWrapper)
+	testFile(t, "tpch_cases.txt", makeTestOutput(t), vschemaWrapper)
 }
 
 func BenchmarkOLTP(b *testing.B) {
@@ -351,10 +325,6 @@ func benchmarkWorkload(b *testing.B, name string) {
 }
 
 func TestBypassPlanningShardTargetFromFile(t *testing.T) {
-	testOutputTempDir, err := os.MkdirTemp("", "plan_test")
-	require.NoError(t, err)
-	defer os.RemoveAll(testOutputTempDir)
-
 	vschema := &vschemaWrapper{
 		v: loadSchema(t, "schema_test.json", true),
 		keyspace: &vindexes.Keyspace{
@@ -364,13 +334,9 @@ func TestBypassPlanningShardTargetFromFile(t *testing.T) {
 		tabletType: topodatapb.TabletType_PRIMARY,
 		dest:       key.DestinationShard("-80")}
 
-	testFile(t, "bypass_shard_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "bypass_shard_cases.txt", makeTestOutput(t), vschema)
 }
 func TestBypassPlanningKeyrangeTargetFromFile(t *testing.T) {
-	testOutputTempDir, err := os.MkdirTemp("", "plan_test")
-	require.NoError(t, err)
-	defer os.RemoveAll(testOutputTempDir)
-
 	keyRange, _ := key.ParseShardingSpec("-")
 
 	vschema := &vschemaWrapper{
@@ -383,18 +349,11 @@ func TestBypassPlanningKeyrangeTargetFromFile(t *testing.T) {
 		dest:       key.DestinationExactKeyRange{KeyRange: keyRange[0]},
 	}
 
-	testFile(t, "bypass_keyrange_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "bypass_keyrange_cases.txt", makeTestOutput(t), vschema)
 }
 
 func TestWithDefaultKeyspaceFromFile(t *testing.T) {
 	// We are testing this separately so we can set a default keyspace
-	testOutputTempDir, err := os.MkdirTemp("", "plan_test")
-	require.NoError(t, err)
-	defer func() {
-		if !t.Failed() {
-			_ = os.RemoveAll(testOutputTempDir)
-		}
-	}()
 	vschema := &vschemaWrapper{
 		v: loadSchema(t, "schema_test.json", true),
 		keyspace: &vindexes.Keyspace{
@@ -404,6 +363,7 @@ func TestWithDefaultKeyspaceFromFile(t *testing.T) {
 		tabletType: topodatapb.TabletType_PRIMARY,
 	}
 
+	testOutputTempDir := makeTestOutput(t)
 	testFile(t, "alterVschema_cases.txt", testOutputTempDir, vschema)
 	testFile(t, "ddl_cases.txt", testOutputTempDir, vschema)
 	testFile(t, "migration_cases.txt", testOutputTempDir, vschema)
@@ -414,23 +374,17 @@ func TestWithDefaultKeyspaceFromFile(t *testing.T) {
 
 func TestWithSystemSchemaAsDefaultKeyspace(t *testing.T) {
 	// We are testing this separately so we can set a default keyspace
-	testOutputTempDir, err := os.MkdirTemp("", "plan_test")
-	require.NoError(t, err)
-	defer os.RemoveAll(testOutputTempDir)
 	vschema := &vschemaWrapper{
 		v:          loadSchema(t, "schema_test.json", true),
 		keyspace:   &vindexes.Keyspace{Name: "information_schema"},
 		tabletType: topodatapb.TabletType_PRIMARY,
 	}
 
-	testFile(t, "sysschema_default.txt", testOutputTempDir, vschema)
+	testFile(t, "sysschema_default.txt", makeTestOutput(t), vschema)
 }
 
 func TestOtherPlanningFromFile(t *testing.T) {
 	// We are testing this separately so we can set a default keyspace
-	testOutputTempDir, err := os.MkdirTemp("", "plan_test")
-	defer os.RemoveAll(testOutputTempDir)
-	require.NoError(t, err)
 	vschema := &vschemaWrapper{
 		v: loadSchema(t, "schema_test.json", true),
 		keyspace: &vindexes.Keyspace{
@@ -440,6 +394,7 @@ func TestOtherPlanningFromFile(t *testing.T) {
 		tabletType: topodatapb.TabletType_PRIMARY,
 	}
 
+	testOutputTempDir := makeTestOutput(t)
 	testFile(t, "other_read_cases.txt", testOutputTempDir, vschema)
 	testFile(t, "other_admin_cases.txt", testOutputTempDir, vschema)
 }
@@ -695,7 +650,6 @@ func testFile(t *testing.T, filename, tempDir string, vschema *vschemaWrapper) {
 		if tempDir != "" {
 			gotFile := fmt.Sprintf("%s/%s", tempDir, filename)
 			_ = os.WriteFile(gotFile, []byte(strings.TrimSpace(expected.String())+"\n"), 0644)
-			fmt.Println(fmt.Sprintf("Errors found in plantests. If the output is correct, run `cp %s/* testdata/` to update test expectations", tempDir)) // nolint
 		}
 	})
 }
diff --git a/go/vt/vtgate/planbuilder/testdata/filter_cases.txt b/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
index 39df69cc86..3b1d22ac6d 100644
--- a/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
@@ -135,7 +135,7 @@ Gen4 plan same as above
     "Query": "select id from `user` where costly = 'aa' and `name` = 'bb'",
     "Table": "`user`",
     "Values": [
-      "VARBINARY(\"bb\")"
+      "VARCHAR(\"bb\")"
     ],
     "Vindex": "name_user_map"
   }
@@ -158,7 +158,7 @@ Gen4 plan same as above
     "Query": "select id from `user` where costly in ('aa', 'bb') and `name` in ::__vals",
     "Table": "`user`",
     "Values": [
-      "(VARBINARY(\"aa\"), VARBINARY(\"bb\"))"
+      "(VARCHAR(\"aa\"), VARCHAR(\"bb\"))"
     ],
     "Vindex": "name_user_map"
   }
@@ -181,7 +181,7 @@ Gen4 plan same as above
     "Query": "select id from `user` where (`name`, col) in (('aa', 'bb'), ('cc', 'dd'))",
     "Table": "`user`",
     "Values": [
-      "(VARBINARY(\"aa\"), VARBINARY(\"cc\"))"
+      "(VARCHAR(\"aa\"), VARCHAR(\"cc\"))"
     ],
     "Vindex": "name_user_map"
   }
@@ -204,7 +204,7 @@ Gen4 plan same as above
     "Query": "select id from `user` where (col, `name`) in (('aa', 'bb'), ('cc', 'dd'))",
     "Table": "`user`",
     "Values": [
-      "(VARBINARY(\"bb\"), VARBINARY(\"dd\"))"
+      "(VARCHAR(\"bb\"), VARCHAR(\"dd\"))"
     ],
     "Vindex": "name_user_map"
   }
@@ -227,7 +227,7 @@ Gen4 plan same as above
     "Query": "select id from `user` where (costly, `name`) in (('aa', 'bb'), ('cc', 'dd'))",
     "Table": "`user`",
     "Values": [
-      "(VARBINARY(\"bb\"), VARBINARY(\"dd\"))"
+      "(VARCHAR(\"bb\"), VARCHAR(\"dd\"))"
     ],
     "Vindex": "name_user_map"
   }
@@ -250,7 +250,7 @@ Gen4 plan same as above
     "Query": "select id from `user` where (`name`, costly) in (('aa', 'bb'), ('cc', 'dd'))",
     "Table": "`user`",
     "Values": [
-      "(VARBINARY(\"aa\"), VARBINARY(\"cc\"))"
+      "(VARCHAR(\"aa\"), VARCHAR(\"cc\"))"
     ],
     "Vindex": "name_user_map"
   }
@@ -273,7 +273,7 @@ Gen4 plan same as above
     "Query": "select id from `user` where (col, costly) in (('aa', 'bb')) and (col, `name`) in (('cc', 'dd'))",
     "Table": "`user`",
     "Values": [
-      "(VARBINARY(\"dd\"))"
+      "(VARCHAR(\"dd\"))"
     ],
     "Vindex": "name_user_map"
   }
@@ -319,7 +319,7 @@ Gen4 plan same as above
     "Query": "select id from `user` where (costly, `name`) in (('aa', 'bb'), ('cc', 'dd'))",
     "Table": "`user`",
     "Values": [
-      "(VARBINARY(\"bb\"), VARBINARY(\"dd\"))"
+      "(VARCHAR(\"bb\"), VARCHAR(\"dd\"))"
     ],
     "Vindex": "name_user_map"
   }
@@ -342,7 +342,7 @@ Gen4 plan same as above
     "Query": "select id from `user` where ((col1, `name`), col2) in ((('aa', 'bb'), 'cc'), (('dd', 'ee'), 'ff'))",
     "Table": "`user`",
     "Values": [
-      "(VARBINARY(\"bb\"), VARBINARY(\"ee\"))"
+      "(VARCHAR(\"bb\"), VARCHAR(\"ee\"))"
     ],
     "Vindex": "name_user_map"
   }
@@ -365,7 +365,7 @@ Gen4 plan same as above
     "Query": "select id from `user` where (`name`, (col1, col2)) in (('aa', ('bb', 'cc')), ('dd', ('ee', 'ff')))",
     "Table": "`user`",
     "Values": [
-      "(VARBINARY(\"aa\"), VARBINARY(\"dd\"))"
+      "(VARCHAR(\"aa\"), VARCHAR(\"dd\"))"
     ],
     "Vindex": "name_user_map"
   }
@@ -996,7 +996,7 @@ Gen4 plan same as above
     "Query": "select id or col as val from `user` where `user`.col = 5 and `user`.id in (1, 2) and `user`.`name` = 'aa'",
     "Table": "`user`",
     "Values": [
-      "VARBINARY(\"aa\")"
+      "VARCHAR(\"aa\")"
     ],
     "Vindex": "name_user_map"
   }
@@ -1019,7 +1019,7 @@ Gen4 plan same as above
     "Query": "select id from `user` where `user`.col = false and `user`.id in (1, 2) and `user`.`name` = 'aa'",
     "Table": "`user`",
     "Values": [
-      "VARBINARY(\"aa\")"
+      "VARCHAR(\"aa\")"
     ],
     "Vindex": "name_user_map"
   }
@@ -1587,7 +1587,7 @@ Gen4 plan same as above
     "Query": "select id from `user` where id = 'aa' and `user`.col in (select user_extra.col from user_extra where user_extra.user_id = 'aa')",
     "Table": "`user`",
     "Values": [
-      "VARBINARY(\"aa\")"
+      "VARCHAR(\"aa\")"
     ],
     "Vindex": "user_index"
   }
@@ -2409,7 +2409,7 @@ Gen4 plan same as above
     "Query": "select c2 from cfc_vindex_col where c1 like 'A%'",
     "Table": "cfc_vindex_col",
     "Values": [
-      "VARBINARY(\"A%\")"
+      "VARCHAR(\"A%\")"
     ],
     "Vindex": "cfc"
   }
@@ -3834,7 +3834,6 @@ Gen4 plan same as above
   }
 }
 
-
 # multi column vindex, partial vindex with SelectEqual
 "select * from multicol_tbl where cola = 1"
 {
diff --git a/go/vt/vtgate/planbuilder/testdata/from_cases.txt b/go/vt/vtgate/planbuilder/testdata/from_cases.txt
index 3aa1b4a314..16ac5f359a 100644
--- a/go/vt/vtgate/planbuilder/testdata/from_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/from_cases.txt
@@ -4477,3 +4477,58 @@ Gen4 plan same as above
     ]
   }
 }
+
+# left join where clauses #2
+"select user.id from user left join user_extra on user.col = user_extra.col where coalesce(user_extra.col, 4) = 5"
+"unsupported: cross-shard left join and where clause"
+{
+  "QueryType": "SELECT",
+  "Original": "select user.id from user left join user_extra on user.col = user_extra.col where coalesce(user_extra.col, 4) = 5",
+  "Instructions": {
+    "OperatorType": "SimpleProjection",
+    "Columns": [
+      1
+    ],
+    "Inputs": [
+      {
+        "OperatorType": "Filter",
+        "Predicate": "coalesce(user_extra.col, 4) = 5",
+        "Inputs": [
+          {
+            "OperatorType": "Join",
+            "Variant": "LeftJoin",
+            "JoinColumnIndexes": "1,-2",
+            "JoinVars": {
+              "user_col": 0
+            },
+            "TableName": "`user`_user_extra",
+            "Inputs": [
+              {
+                "OperatorType": "Route",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FieldQuery": "select `user`.col, `user`.id from `user` where 1 != 1",
+                "Query": "select `user`.col, `user`.id from `user`",
+                "Table": "`user`"
+              },
+              {
+                "OperatorType": "Route",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FieldQuery": "select user_extra.col from user_extra where 1 != 1",
+                "Query": "select user_extra.col from user_extra where user_extra.col = :user_col",
+                "Table": "user_extra"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  }
+}
diff --git a/go/vt/vtgate/planbuilder/testdata/select_cases.txt b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
index a156588013..e354223570 100644
--- a/go/vt/vtgate/planbuilder/testdata/select_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
@@ -626,7 +626,7 @@ Gen4 plan same as above
     "Query": "select * from pin_test",
     "Table": "pin_test",
     "Values": [
-      "VARBINARY(\"\\x80\")"
+      "VARCHAR(\"\\x80\")"
     ],
     "Vindex": "binary"
   }
diff --git a/go/vt/vtgate/planbuilder/testdata/sysschema_default.txt b/go/vt/vtgate/planbuilder/testdata/sysschema_default.txt
index 7e9e288cc7..83e05b9312 100644
--- a/go/vt/vtgate/planbuilder/testdata/sysschema_default.txt
+++ b/go/vt/vtgate/planbuilder/testdata/sysschema_default.txt
@@ -31,7 +31,7 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableSchema": "[VARCHAR(\"user\"), VARCHAR(\"user\")]",
     "Table": "information_schema.`tables`, information_schema.`columns`"
   }
 }
@@ -47,7 +47,7 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableSchema": "[VARCHAR(\"user\"), VARCHAR(\"user\")]",
     "Table": "information_schema.`columns`, information_schema.`tables`"
   }
 }
@@ -66,7 +66,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select (select 1 from information_schema.schemata where 1 != 1) from dual where 1 != 1",
     "Query": "select (select 1 from information_schema.schemata where schema_name = :__vtschemaname limit 1) from dual",
-    "SysTableTableSchema": "[VARBINARY(\"MyDatabase\")]",
+    "SysTableTableSchema": "[VARCHAR(\"MyDatabase\")]",
     "Table": "dual"
   }
 }
@@ -86,7 +86,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select * from (select 1 from information_schema.schemata where 1 != 1) as x where 1 != 1",
     "Query": "select * from (select 1 from information_schema.schemata where schema_name = :__vtschemaname limit 1) as x",
-    "SysTableTableSchema": "[VARBINARY(\"MyDatabase\")]",
+    "SysTableTableSchema": "[VARCHAR(\"MyDatabase\")]",
     "Table": "information_schema.schemata"
   }
 }
@@ -102,7 +102,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select x.`1` from (select 1 from information_schema.schemata where 1 != 1) as x where 1 != 1",
     "Query": "select x.`1` from (select 1 from information_schema.schemata where schema_name = :__vtschemaname limit 1) as x",
-    "SysTableTableSchema": "[VARBINARY(\"MyDatabase\")]",
+    "SysTableTableSchema": "[VARCHAR(\"MyDatabase\")]",
     "Table": "information_schema.schemata"
   }
 }
diff --git a/go/vt/vtgate/planbuilder/testdata/systemtables_cases.txt b/go/vt/vtgate/planbuilder/testdata/systemtables_cases.txt
index 075ccaebb1..5f5da81c80 100644
--- a/go/vt/vtgate/planbuilder/testdata/systemtables_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/systemtables_cases.txt
@@ -179,7 +179,7 @@ Gen4 plan same as above
             },
             "FieldQuery": "select * from information_schema.`tables` where 1 != 1",
             "Query": "select * from information_schema.`tables` where table_schema = :__vtschemaname",
-            "SysTableTableSchema": "[VARBINARY(\"user\")]",
+            "SysTableTableSchema": "[VARCHAR(\"user\")]",
             "Table": "information_schema.`tables`"
           },
           {
@@ -191,7 +191,7 @@ Gen4 plan same as above
             },
             "FieldQuery": "select * from information_schema.`tables` where 1 != 1",
             "Query": "select * from information_schema.`tables` where table_schema = :__vtschemaname",
-            "SysTableTableSchema": "[VARBINARY(\"main\")]",
+            "SysTableTableSchema": "[VARCHAR(\"main\")]",
             "Table": "information_schema.`tables`"
           }
         ]
@@ -217,7 +217,7 @@ Gen4 plan same as above
             },
             "FieldQuery": "select * from information_schema.`tables` where 1 != 1",
             "Query": "select distinct * from information_schema.`tables` where table_schema = :__vtschemaname",
-            "SysTableTableSchema": "[VARBINARY(\"user\")]",
+            "SysTableTableSchema": "[VARCHAR(\"user\")]",
             "Table": "information_schema.`tables`"
           },
           {
@@ -229,7 +229,7 @@ Gen4 plan same as above
             },
             "FieldQuery": "select * from information_schema.`tables` where 1 != 1",
             "Query": "select distinct * from information_schema.`tables` where table_schema = :__vtschemaname",
-            "SysTableTableSchema": "[VARBINARY(\"main\")]",
+            "SysTableTableSchema": "[VARCHAR(\"main\")]",
             "Table": "information_schema.`tables`"
           }
         ]
@@ -252,8 +252,8 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[KCU_TABLE_NAME:VARCHAR(\"data_type_table\")]",
+    "SysTableTableSchema": "[VARCHAR(\"test\")]",
     "Table": "INFORMATION_SCHEMA.KEY_COLUMN_USAGE, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS"
   }
 }
@@ -269,8 +269,8 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[KCU_TABLE_NAME:VARCHAR(\"data_type_table\")]",
+    "SysTableTableSchema": "[VARCHAR(\"test\")]",
     "Table": "INFORMATION_SCHEMA.KEY_COLUMN_USAGE, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS"
   }
 }
@@ -295,8 +295,8 @@ Gen4 plan same as above
         },
         "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_NAME1 order by KCU.CONSTRAINT_NAME asc, KCU.COLUMN_NAME asc",
-        "SysTableTableName": "[KCU_TABLE_NAME1:VARBINARY(\"data_type_table\"), KCU_TABLE_NAME:VARBINARY(\"data_type_table\")]",
-        "SysTableTableSchema": "[VARBINARY(\"test\")]",
+        "SysTableTableName": "[KCU_TABLE_NAME1:VARCHAR(\"data_type_table\"), KCU_TABLE_NAME:VARCHAR(\"data_type_table\")]",
+        "SysTableTableSchema": "[VARCHAR(\"test\")]",
         "Table": "INFORMATION_SCHEMA.KEY_COLUMN_USAGE, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS"
       },
       {
@@ -308,8 +308,8 @@ Gen4 plan same as above
         },
         "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\")]",
+        "SysTableTableName": "[S_TABLE_NAME:VARCHAR(\"sc\")]",
+        "SysTableTableSchema": "[VARCHAR(\"test\")]",
         "Table": "INFORMATION_SCHEMA.K"
       }
     ]
@@ -327,8 +327,8 @@ Gen4 plan same as above
     },
     "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.TABLE_NAME = :KCU_TABLE_NAME1 and KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME order by KCU.CONSTRAINT_NAME asc, KCU.COLUMN_NAME asc",
-    "SysTableTableName": "[KCU_TABLE_NAME1:VARBINARY(\"data_type_table\"), KCU_TABLE_NAME:VARBINARY(\"data_type_table\"), S_TABLE_NAME:VARBINARY(\"sc\")]",
-    "SysTableTableSchema": "[VARBINARY(\"test\"), VARBINARY(\"test\")]",
+    "SysTableTableName": "[KCU_TABLE_NAME1:VARCHAR(\"data_type_table\"), KCU_TABLE_NAME:VARCHAR(\"data_type_table\"), S_TABLE_NAME:VARCHAR(\"sc\")]",
+    "SysTableTableSchema": "[VARCHAR(\"test\"), VARCHAR(\"test\")]",
     "Table": "INFORMATION_SCHEMA.K, INFORMATION_SCHEMA.KEY_COLUMN_USAGE, INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS"
   }
 }
@@ -422,7 +422,7 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[fk_table_name:VARCHAR(\":vtg1\"), rc_table_name:VARCHAR(\":vtg1\")]",
     "Table": "information_schema.referential_constraints, information_schema.key_column_usage"
   }
 }
@@ -438,7 +438,7 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[fk_table_name:VARCHAR(\":vtg1\"), rc_table_name:VARCHAR(\":vtg1\")]",
     "Table": "information_schema.key_column_usage, information_schema.referential_constraints"
   }
 }
@@ -457,7 +457,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select * from information_schema.schemata where 1 != 1",
     "Query": "select * from information_schema.schemata where schema_name = :__vtschemaname",
-    "SysTableTableSchema": "[VARBINARY(\"user\")]",
+    "SysTableTableSchema": "[VARCHAR(\"user\")]",
     "Table": "information_schema.schemata"
   }
 }
@@ -477,8 +477,8 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[table_name:VARCHAR(\"table_name\")]",
+    "SysTableTableSchema": "[VARCHAR(\"schema_name\")]",
     "Table": "information_schema.`tables`"
   }
 }
@@ -498,8 +498,8 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[fk_table_name:VARCHAR(\"table_name\"), rc_table_name:VARCHAR(\"table_name\")]",
+    "SysTableTableSchema": "[VARCHAR(\"table_schema\"), VARCHAR(\"table_schema\")]",
     "Table": "information_schema.referential_constraints, information_schema.key_column_usage"
   }
 }
@@ -515,8 +515,8 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[fk_table_name:VARCHAR(\"table_name\"), rc_table_name:VARCHAR(\"table_name\")]",
+    "SysTableTableSchema": "[VARCHAR(\"table_schema\"), VARCHAR(\"table_schema\")]",
     "Table": "information_schema.key_column_usage, information_schema.referential_constraints"
   }
 }
@@ -535,8 +535,8 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[tc_table_name:VARCHAR(\"table_name\")]",
+    "SysTableTableSchema": "[VARCHAR(\"table_schema\"), VARCHAR(\"constraint_schema\")]",
     "Table": "information_schema.check_constraints, information_schema.table_constraints"
   }
 }
@@ -552,8 +552,8 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[tc_table_name:VARCHAR(\"table_name\")]",
+    "SysTableTableSchema": "[VARCHAR(\"constraint_schema\"), VARCHAR(\"table_schema\")]",
     "Table": "information_schema.check_constraints, information_schema.table_constraints"
   }
 }
@@ -572,8 +572,8 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[table_name:VARCHAR(\"table_name\")]",
+    "SysTableTableSchema": "[VARCHAR(\"table_schema\")]",
     "Table": "information_schema.statistics"
   }
 }
@@ -593,8 +593,8 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[table_name:VARCHAR(\"table_name\")]",
+    "SysTableTableSchema": "[VARCHAR(\"table_schema\")]",
     "Table": "information_schema.`columns`"
   }
 }
@@ -633,7 +633,7 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableSchema": "[VARCHAR(\"table_schema\")]",
     "Table": "information_schema.`tables`"
   }
 }
@@ -653,8 +653,8 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[_subquery_table_name:VARCHAR(\"table_name\")]",
+    "SysTableTableSchema": "[VARCHAR(\"table_schema\")]",
     "Table": "information_schema.`tables`"
   }
 }
@@ -670,7 +670,7 @@ Gen4 plan same as above
     },
     "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 and table_type = 'table_type' and table_name = 'table_name') as _subquery",
-    "SysTableTableSchema": "[VARBINARY(\"table_schema\")]",
+    "SysTableTableSchema": "[VARCHAR(\"table_schema\")]",
     "Table": "information_schema.`tables`"
   }
 }
@@ -689,7 +689,7 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableSchema": "[VARCHAR(\"a\"), VARCHAR(\"a\")]",
     "Table": "information_schema.check_constraints"
   }
 }
@@ -709,8 +709,8 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[table_name:VARCHAR(\"foo\")]",
+    "SysTableTableSchema": "[VARCHAR(\"performance_schema\")]",
     "Table": "INFORMATION_SCHEMA.`TABLES`"
   }
 }
@@ -749,7 +749,7 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableSchema": "[VARCHAR(\"user\"), VARCHAR(\"main\")]",
     "Table": "INFORMATION_SCHEMA.`TABLES`"
   }
 }
@@ -788,7 +788,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select * from INFORMATION_SCHEMA.`TABLES` where 1 != 1",
     "Query": "select * from INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = :__vtschemaname",
-    "SysTableTableSchema": "[VARBINARY(\"ks\")]",
+    "SysTableTableSchema": "[VARCHAR(\"ks\")]",
     "Table": "INFORMATION_SCHEMA.`TABLES`"
   }
 }
@@ -808,8 +808,8 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableName": "[TABLE_NAME:VARCHAR(\"route1\")]",
+    "SysTableTableSchema": "[VARCHAR(\"ks\")]",
     "Table": "INFORMATION_SCHEMA.`TABLES`"
   }
 }
@@ -829,7 +829,7 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableSchema": "[VARCHAR(\"ks\")]",
     "Table": "INFORMATION_SCHEMA.`TABLES`"
   }
 }
@@ -849,7 +849,7 @@ Gen4 plan same as above
     },
     "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\")]",
+    "SysTableTableSchema": "[VARCHAR(\"ks\")]",
     "Table": "INFORMATION_SCHEMA.`TABLES`"
   }
 }
@@ -992,7 +992,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select a.table_name from (select * from information_schema.key_column_usage as a where 1 != 1) as a join (select * from information_schema.referential_constraints where 1 != 1) as b where 1 != 1",
     "Query": "select a.table_name from (select * from information_schema.key_column_usage as a where a.table_name = :a_table_name) as a join (select * from information_schema.referential_constraints where table_name = :table_name) as b",
-    "SysTableTableName": "[a_table_name:VARBINARY(\"users\"), table_name:VARBINARY(\"users\")]",
+    "SysTableTableName": "[a_table_name:VARCHAR(\"users\"), table_name:VARCHAR(\"users\")]",
     "Table": "information_schema.key_column_usage, information_schema.referential_constraints"
   }
 }
@@ -1008,7 +1008,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select a.table_name from (select * from information_schema.key_column_usage as a where 1 != 1) as a, (select * from information_schema.referential_constraints where 1 != 1) as b where 1 != 1",
     "Query": "select a.table_name from (select * from information_schema.key_column_usage as a where a.table_name = :a_table_name) as a, (select * from information_schema.referential_constraints where table_name = :table_name) as b",
-    "SysTableTableName": "[a_table_name:VARBINARY(\"users\"), table_name:VARBINARY(\"users\")]",
+    "SysTableTableName": "[a_table_name:VARCHAR(\"users\"), table_name:VARCHAR(\"users\")]",
     "Table": "information_schema.key_column_usage, information_schema.referential_constraints"
   }
 }
@@ -1027,7 +1027,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select sum(found) from (select 1 as found from information_schema.`tables` where 1 != 1 union all (select 1 as found from information_schema.views where 1 != 1)) as t where 1 != 1",
     "Query": "select sum(found) from (select 1 as found from information_schema.`tables` where table_schema = :__vtschemaname union all (select 1 as found from information_schema.views where table_schema = :__vtschemaname limit 1)) as t",
-    "SysTableTableSchema": "[VARBINARY(\"music\"), VARBINARY(\"music\")]",
+    "SysTableTableSchema": "[VARCHAR(\"music\"), VARCHAR(\"music\")]",
     "Table": "information_schema.`tables`"
   }
 }
@@ -1055,7 +1055,7 @@ Gen4 plan same as above
             },
             "FieldQuery": "select 1 as found from information_schema.`tables` where 1 != 1",
             "Query": "select 1 as found from information_schema.`tables` where table_schema = :__vtschemaname",
-            "SysTableTableSchema": "[VARBINARY(\"music\")]",
+            "SysTableTableSchema": "[VARCHAR(\"music\")]",
             "Table": "information_schema.`tables`"
           },
           {
@@ -1067,7 +1067,7 @@ Gen4 plan same as above
             },
             "FieldQuery": "select 1 as found from information_schema.views where 1 != 1",
             "Query": "select 1 as found from information_schema.views where table_schema = :__vtschemaname limit 1",
-            "SysTableTableSchema": "[VARBINARY(\"music\")]",
+            "SysTableTableSchema": "[VARCHAR(\"music\")]",
             "Table": "information_schema.views"
           }
         ]
@@ -1087,7 +1087,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select found from (select 1 as found from information_schema.`tables` where 1 != 1 union all (select 1 as found from information_schema.views where 1 != 1)) as t where 1 != 1",
     "Query": "select found from (select 1 as found from information_schema.`tables` where table_schema = :__vtschemaname union all (select 1 as found from information_schema.views where table_schema = :__vtschemaname limit 1)) as t",
-    "SysTableTableSchema": "[VARBINARY(\"music\"), VARBINARY(\"music\")]",
+    "SysTableTableSchema": "[VARCHAR(\"music\"), VARCHAR(\"music\")]",
     "Table": "information_schema.`tables`"
   }
 }
@@ -1109,7 +1109,7 @@ Gen4 plan same as above
         },
         "FieldQuery": "select 1 as found from information_schema.`tables` where 1 != 1",
         "Query": "select 1 as found from information_schema.`tables` where table_schema = :__vtschemaname and table_schema = :__vtschemaname",
-        "SysTableTableSchema": "[VARBINARY(\"music\"), VARBINARY(\"Music\")]",
+        "SysTableTableSchema": "[VARCHAR(\"music\"), VARCHAR(\"Music\")]",
         "Table": "information_schema.`tables`"
       },
       {
@@ -1121,7 +1121,7 @@ Gen4 plan same as above
         },
         "FieldQuery": "select 1 as found from information_schema.views where 1 != 1",
         "Query": "select 1 as found from information_schema.views where table_schema = :__vtschemaname and table_schema = :__vtschemaname limit 1",
-        "SysTableTableSchema": "[VARBINARY(\"music\"), VARBINARY(\"user\")]",
+        "SysTableTableSchema": "[VARCHAR(\"music\"), VARCHAR(\"user\")]",
         "Table": "information_schema.views"
       }
     ]
@@ -1139,7 +1139,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select 1 as found from information_schema.`tables` where 1 != 1 union all (select 1 as found from information_schema.views where 1 != 1)",
     "Query": "select 1 as found from information_schema.`tables` where table_schema = :__vtschemaname and table_schema = :__vtschemaname union all (select 1 as found from information_schema.views where table_schema = :__vtschemaname and table_schema = :__vtschemaname limit 1)",
-    "SysTableTableSchema": "[VARBINARY(\"music\"), VARBINARY(\"Music\"), VARBINARY(\"music\"), VARBINARY(\"user\")]",
+    "SysTableTableSchema": "[VARCHAR(\"music\"), VARCHAR(\"Music\"), VARCHAR(\"music\"), VARCHAR(\"user\")]",
     "Table": "information_schema.`tables`"
   }
 }
@@ -1161,7 +1161,7 @@ Gen4 plan same as above
         },
         "FieldQuery": "select 1 as found from information_schema.`tables` where 1 != 1",
         "Query": "select 1 as found from information_schema.`tables` where table_schema = :__vtschemaname and table_schema = :__vtschemaname",
-        "SysTableTableSchema": "[VARBINARY(\"music\"), VARBINARY(\"Music\")]",
+        "SysTableTableSchema": "[VARCHAR(\"music\"), VARCHAR(\"Music\")]",
         "Table": "information_schema.`tables`"
       },
       {
@@ -1173,7 +1173,7 @@ Gen4 plan same as above
         },
         "FieldQuery": "select 1 as found from information_schema.views where 1 != 1",
         "Query": "select 1 as found from information_schema.views where table_schema = :__vtschemaname and table_schema = :__vtschemaname limit 1",
-        "SysTableTableSchema": "[VARBINARY(\"music\"), VARBINARY(\"user\")]",
+        "SysTableTableSchema": "[VARCHAR(\"music\"), VARCHAR(\"user\")]",
         "Table": "information_schema.views"
       }
     ]
@@ -1191,7 +1191,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select 1 as found from information_schema.`tables` where 1 != 1 union all (select 1 as found from information_schema.views where 1 != 1)",
     "Query": "select 1 as found from information_schema.`tables` where table_schema = :__vtschemaname and table_schema = :__vtschemaname union all (select 1 as found from information_schema.views where table_schema = :__vtschemaname and table_schema = :__vtschemaname limit 1)",
-    "SysTableTableSchema": "[VARBINARY(\"music\"), VARBINARY(\"Music\"), VARBINARY(\"music\"), VARBINARY(\"user\")]",
+    "SysTableTableSchema": "[VARCHAR(\"music\"), VARCHAR(\"Music\"), VARCHAR(\"music\"), VARCHAR(\"user\")]",
     "Table": "information_schema.`tables`"
   }
 }
@@ -1221,7 +1221,7 @@ Gen4 plan same as above
             },
             "FieldQuery": "select 1 as found from information_schema.`tables` where 1 != 1",
             "Query": "select 1 as found from information_schema.`tables` where table_name = :table_name2 and table_name = :table_name3",
-            "SysTableTableName": "[table_name2:VARBINARY(\"music\"), table_name3:VARBINARY(\"Music\")]",
+            "SysTableTableName": "[table_name2:VARCHAR(\"music\"), table_name3:VARCHAR(\"Music\")]",
             "Table": "information_schema.`tables`"
           },
           {
@@ -1233,7 +1233,7 @@ Gen4 plan same as above
             },
             "FieldQuery": "select 1 as found from information_schema.views where 1 != 1",
             "Query": "select 1 as found from information_schema.views where table_name = :table_name4 and table_name = :table_name5 limit 1",
-            "SysTableTableName": "[table_name4:VARBINARY(\"music\"), table_name5:VARBINARY(\"user\")]",
+            "SysTableTableName": "[table_name4:VARCHAR(\"music\"), table_name5:VARCHAR(\"user\")]",
             "Table": "information_schema.views"
           }
         ]
@@ -1247,7 +1247,7 @@ Gen4 plan same as above
         },
         "FieldQuery": "select 1 as found from information_schema.`tables` where 1 != 1",
         "Query": "select 1 as found from information_schema.`tables` where table_name = :table_name and table_name = :table_name1 and :__sq_has_values1",
-        "SysTableTableName": "[table_name1:VARBINARY(\"Music\"), table_name:VARBINARY(\"music\")]",
+        "SysTableTableName": "[table_name1:VARCHAR(\"Music\"), table_name:VARCHAR(\"music\")]",
         "Table": "information_schema.`tables`"
       }
     ]
@@ -1265,7 +1265,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select 1 as found from information_schema.`tables` where 1 != 1",
     "Query": "select 1 as found from information_schema.`tables` where table_name = :table_name and table_name = :table_name1 and exists (select 1 as found from information_schema.`tables` where table_name = :table_name2 and table_name = :table_name3 union all (select 1 as found from information_schema.views where table_name = :table_name4 and table_name = :table_name5 limit 1))",
-    "SysTableTableName": "[table_name1:VARBINARY(\"Music\"), table_name2:VARBINARY(\"music\"), table_name3:VARBINARY(\"Music\"), table_name4:VARBINARY(\"music\"), table_name5:VARBINARY(\"user\"), table_name:VARBINARY(\"music\")]",
+    "SysTableTableName": "[table_name1:VARCHAR(\"Music\"), table_name2:VARCHAR(\"music\"), table_name3:VARCHAR(\"Music\"), table_name4:VARCHAR(\"music\"), table_name5:VARCHAR(\"user\"), table_name:VARCHAR(\"music\")]",
     "Table": "information_schema.`tables`"
   }
 }
@@ -1296,7 +1296,7 @@ Gen4 plan same as above
                 },
                 "FieldQuery": "select id from information_schema.`table` as t where 1 != 1",
                 "Query": "select id from information_schema.`table` as t where t.schema_name = :__vtschemaname",
-                "SysTableTableSchema": "[VARBINARY(\"a\")]",
+                "SysTableTableSchema": "[VARCHAR(\"a\")]",
                 "Table": "information_schema.`table`"
               },
               {
@@ -1329,7 +1329,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select id from (select id from information_schema.`table` as t where 1 != 1 union select id from information_schema.`columns` where 1 != 1) as dt where 1 != 1",
     "Query": "select id from (select id from information_schema.`table` as t where t.schema_name = :__vtschemaname union select id from information_schema.`columns`) as dt",
-    "SysTableTableSchema": "[VARBINARY(\"a\")]",
+    "SysTableTableSchema": "[VARCHAR(\"a\")]",
     "Table": "information_schema.`table`"
   }
 }
@@ -1362,7 +1362,7 @@ Gen4 plan same as above
                 },
                 "FieldQuery": "select id from information_schema.`table` as t where 1 != 1",
                 "Query": "select id from information_schema.`table` as t where t.schema_name = :__vtschemaname",
-                "SysTableTableSchema": "[VARBINARY(\"a\")]",
+                "SysTableTableSchema": "[VARCHAR(\"a\")]",
                 "Table": "information_schema.`table`"
               },
               {
@@ -1406,7 +1406,7 @@ Gen4 plan same as above
     },
     "FieldQuery": "select id from information_schema.random as t where 1 != 1",
     "Query": "select id from information_schema.random as t where t.col in (select id from information_schema.`table` as t where t.schema_name = :__vtschemaname union select id from information_schema.`columns`)",
-    "SysTableTableSchema": "[VARBINARY(\"a\")]",
+    "SysTableTableSchema": "[VARCHAR(\"a\")]",
     "Table": "information_schema.random"
   }
 }
@@ -1426,8 +1426,8 @@ Gen4 plan same as above
     },
     "FieldQuery": "select * from (select * from information_schema.key_column_usage as kcu where 1 != 1 union select * from information_schema.key_column_usage as kcu where 1 != 1) as kcu where 1 != 1",
     "Query": "select * from (select * from information_schema.key_column_usage as kcu where kcu.table_schema = :__vtschemaname and kcu.table_name = :kcu_table_name and constraint_name = 'primary' union select * from information_schema.key_column_usage as kcu where kcu.table_schema = :__vtschemaname and kcu.table_name = :kcu_table_name1 and constraint_name = 'primary') as kcu",
-    "SysTableTableName": "[kcu_table_name1:VARBINARY(\"music\"), kcu_table_name:VARBINARY(\"user_extra\")]",
-    "SysTableTableSchema": "[VARBINARY(\"user\"), VARBINARY(\"user\")]",
+    "SysTableTableName": "[kcu_table_name1:VARCHAR(\"music\"), kcu_table_name:VARCHAR(\"user_extra\")]",
+    "SysTableTableSchema": "[VARCHAR(\"user\"), VARCHAR(\"user\")]",
     "Table": "information_schema.key_column_usage"
   }
 }
diff --git a/go/vt/vtgate/planbuilder/testdata/union_cases.txt b/go/vt/vtgate/planbuilder/testdata/union_cases.txt
index f1cf39f80c..77f5c033ec 100644
--- a/go/vt/vtgate/planbuilder/testdata/union_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/union_cases.txt
@@ -1773,88 +1773,7 @@ Gen4 plan same as above
     ]
   }
 }
-{
-  "QueryType": "SELECT",
-  "Original": "select tbl2.id FROM ((select id from user order by id limit 5) union all (select id from user order by id desc limit 5)) as tbl1 INNER JOIN user as tbl2  ON tbl1.id = tbl2.id",
-  "Instructions": {
-    "OperatorType": "Join",
-    "Variant": "Join",
-    "JoinColumnIndexes": "1",
-    "JoinVars": {
-      "tbl1_id": 0
-    },
-    "TableName": "`user`_`user`",
-    "Inputs": [
-      {
-        "OperatorType": "SimpleProjection",
-        "Columns": [
-          0
-        ],
-        "Inputs": [
-          {
-            "OperatorType": "Concatenate",
-            "Inputs": [
-              {
-                "OperatorType": "Limit",
-                "Count": "INT64(5)",
-                "Inputs": [
-                  {
-                    "OperatorType": "Route",
-                    "Variant": "Scatter",
-                    "Keyspace": {
-                      "Name": "user",
-                      "Sharded": true
-                    },
-                    "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1",
-                    "OrderBy": "(0|1) ASC",
-                    "Query": "select id, weight_string(id) from `user` order by id asc limit :__upper_limit",
-                    "ResultColumns": 1,
-                    "Table": "`user`"
-                  }
-                ]
-              },
-              {
-                "OperatorType": "Limit",
-                "Count": "INT64(5)",
-                "Inputs": [
-                  {
-                    "OperatorType": "Route",
-                    "Variant": "Scatter",
-                    "Keyspace": {
-                      "Name": "user",
-                      "Sharded": true
-                    },
-                    "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1",
-                    "OrderBy": "(0|1) DESC",
-                    "Query": "select id, weight_string(id) from `user` order by id desc limit :__upper_limit",
-                    "ResultColumns": 1,
-                    "Table": "`user`"
-                  }
-                ]
-              }
-            ]
-          }
-        ]
-      },
-      {
-        "OperatorType": "Route",
-        "Variant": "EqualUnique",
-        "Keyspace": {
-          "Name": "user",
-          "Sharded": true
-        },
-        "FieldQuery": "select tbl2.id from `user` as tbl2 where 1 != 1",
-        "Query": "select tbl2.id from `user` as tbl2 where tbl2.id = :tbl1_id",
-        "Table": "`user`",
-        "Values": [
-          ":tbl1_id"
-        ],
-        "Vindex": "user_index"
-      }
-    ]
-  }
-}
-
+Gen4 plan same as above
 
 # ambiguous LIMIT
 "select id from user limit 1 union all select id from music limit 1"
diff --git a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
index 62320cbb1a..e048f8832a 100644
--- a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
@@ -68,11 +68,6 @@ Gen4 plan same as above
 "unsupported: cross-shard left join and column expressions"
 Gen4 plan same as above
 
-# left join where clauses #2
-"select user.id from user left join user_extra on user.col = user_extra.col where coalesce(user_extra.col, 4) = 5"
-"unsupported: cross-shard left join and where clause"
-Gen4 error: expr cannot be converted, not supported: coalesce(user_extra.col, 4) = 5
-
 # * expresson not allowed for cross-shard joins
 "select * from user join user_extra"
 "unsupported: '*' expression in cross-shard query"
@@ -551,7 +546,7 @@ Gen4 error: expr cannot be converted, not supported: count(id) = 10
     "Query": "select 1 from `user` having count(id) = 10 and `name` = 'a'",
     "Table": "`user`",
     "Values": [
-      "VARBINARY(\"a\")"
+      "VARCHAR(\"a\")"
     ],
     "Vindex": "name_user_map"
   }
diff --git a/go/vt/vttablet/endtoend/misc_test.go b/go/vt/vttablet/endtoend/misc_test.go
index 0fece95690..264979ceeb 100644
--- a/go/vt/vttablet/endtoend/misc_test.go
+++ b/go/vt/vttablet/endtoend/misc_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package endtoend
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"net/http"
@@ -30,8 +31,6 @@ import (
 
 	"vitess.io/vitess/go/test/utils"
 
-	"context"
-
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -571,7 +570,7 @@ func TestLogTruncation(t *testing.T) {
 		"insert into vitess_test values(123, null, :data, null)",
 		map[string]*querypb.BindVariable{"data": sqltypes.StringBindVariable("THIS IS A LONG LONG LONG LONG QUERY STRING THAT SHOULD BE SHORTENED")},
 	)
-	wantLog := `Data too long for column 'charval' at row 1 (errno 1406) (sqlstate 22001) (CallerID: dev): Sql: "insert into vitess_test values(123, null, :data, null)", BindVars: {data: "type:VARBINARY value:\"THIS IS A LONG LONG LONG LONG QUERY STRING THAT SHOULD BE SHORTENED\""}`
+	wantLog := `Data too long for column 'charval' at row 1 (errno 1406) (sqlstate 22001) (CallerID: dev): Sql: "insert into vitess_test values(123, null, :data, null)", BindVars: {data: "type:VARCHAR value:\"THIS IS A LONG LONG LONG LONG QUERY STRING THAT SHOULD BE SHORTENED\""}`
 	wantErr := wantLog
 	if err == nil {
 		t.Errorf("query unexpectedly succeeded")
@@ -591,7 +590,7 @@ func TestLogTruncation(t *testing.T) {
 		map[string]*querypb.BindVariable{"data": sqltypes.StringBindVariable("THIS IS A LONG LONG LONG LONG QUERY STRING THAT SHOULD BE SHORTENED")},
 	)
 	wantLog = `Data too long for column 'charval' at row 1 (errno 1406) (sqlstate 22001) (CallerID: dev): Sql: "insert into vitess [TRUNCATED]", BindVars: {data: " [TRUNCATED]`
-	wantErr = `Data too long for column 'charval' at row 1 (errno 1406) (sqlstate 22001) (CallerID: dev): Sql: "insert into vitess_test values(123, null, :data, null)", BindVars: {data: "type:VARBINARY value:\"THIS IS A LONG LONG LONG LONG QUERY STRING THAT SHOULD BE SHORTENED\""}`
+	wantErr = `Data too long for column 'charval' at row 1 (errno 1406) (sqlstate 22001) (CallerID: dev): Sql: "insert into vitess_test values(123, null, :data, null)", BindVars: {data: "type:VARCHAR value:\"THIS IS A LONG LONG LONG LONG QUERY STRING THAT SHOULD BE SHORTENED\""}`
 	if err == nil {
 		t.Errorf("query unexpectedly succeeded")
 	}
@@ -609,7 +608,7 @@ func TestLogTruncation(t *testing.T) {
 		map[string]*querypb.BindVariable{"data": sqltypes.StringBindVariable("THIS IS A LONG LONG LONG LONG QUERY STRING THAT SHOULD BE SHORTENED")},
 	)
 	wantLog = `Data too long for column 'charval' at row 1 (errno 1406) (sqlstate 22001) (CallerID: dev): Sql: "insert into vitess [TRUNCATED] /* KEEP ME */", BindVars: {data: " [TRUNCATED]`
-	wantErr = `Data too long for column 'charval' at row 1 (errno 1406) (sqlstate 22001) (CallerID: dev): Sql: "insert into vitess_test values(123, null, :data, null) /* KEEP ME */", BindVars: {data: "type:VARBINARY value:\"THIS IS A LONG LONG LONG LONG QUERY STRING THAT SHOULD BE SHORTENED\""}`
+	wantErr = `Data too long for column 'charval' at row 1 (errno 1406) (sqlstate 22001) (CallerID: dev): Sql: "insert into vitess_test values(123, null, :data, null) /* KEEP ME */", BindVars: {data: "type:VARCHAR value:\"THIS IS A LONG LONG LONG LONG QUERY STRING THAT SHOULD BE SHORTENED\""}`
 	if err == nil {
 		t.Errorf("query unexpectedly succeeded")
 	}
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vplayer_flaky_test.go b/go/vt/vttablet/tabletmanager/vreplication/vplayer_flaky_test.go
index db8ed3e0ee..1979e5ad38 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vplayer_flaky_test.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vplayer_flaky_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package vreplication
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	"math"
@@ -27,18 +28,13 @@ import (
 	"testing"
 	"time"
 
-	"vitess.io/vitess/go/mysql"
-
 	"github.com/spyzhov/ajson"
 	"github.com/stretchr/testify/require"
 
-	"vitess.io/vitess/go/vt/log"
-
-	"context"
-
+	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/binlog/binlogplayer"
-
+	"vitess.io/vitess/go/vt/log"
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
 )
 
diff --git a/go/vt/vttablet/tabletserver/messager/message_manager_test.go b/go/vt/vttablet/tabletserver/messager/message_manager_test.go
index 78b2d49798..7006ff7869 100644
--- a/go/vt/vttablet/tabletserver/messager/message_manager_test.go
+++ b/go/vt/vttablet/tabletserver/messager/message_manager_test.go
@@ -749,7 +749,7 @@ func TestMMGenerate(t *testing.T) {
 		t.Errorf("gotAcked: %d, should be with 10s of %d", gotAcked, wantAcked)
 	}
 	gotids := bv["ids"]
-	wantids := sqltypes.TestBindVariable([]interface{}{"1", "2"})
+	wantids := sqltypes.TestBindVariable([]interface{}{[]byte{'1'}, []byte{'2'}})
 	utils.MustMatch(t, wantids, gotids, "did not match")
 
 	query, bv = mm.GeneratePostponeQuery([]string{"1", "2"})
@@ -794,7 +794,7 @@ func TestMMGenerateWithBackoff(t *testing.T) {
 	mm.Open()
 	defer mm.Close()
 
-	wantids := sqltypes.TestBindVariable([]interface{}{"1", "2"})
+	wantids := sqltypes.TestBindVariable([]interface{}{[]byte{'1'}, []byte{'2'}})
 
 	query, bv := mm.GeneratePostponeQuery([]string{"1", "2"})
 	wantQuery := "update foo set time_next = :time_now + :wait_time + IF(FLOOR((:min_backoff<<ifnull(epoch, 0)) * :jitter) < :min_backoff, :min_backoff, IF(FLOOR((:min_backoff<<ifnull(epoch, 0)) * :jitter) > :max_backoff, :max_backoff, FLOOR((:min_backoff<<ifnull(epoch, 0)) * :jitter))), epoch = ifnull(epoch, 0)+1 where id in ::ids and time_acked is null"
diff --git a/go/vt/vttablet/tabletserver/tabletenv/logstats_test.go b/go/vt/vttablet/tabletserver/tabletenv/logstats_test.go
index bf471ae5e4..25ace3200c 100644
--- a/go/vt/vttablet/tabletserver/tabletenv/logstats_test.go
+++ b/go/vt/vttablet/tabletserver/tabletenv/logstats_test.go
@@ -18,6 +18,7 @@ package tabletenv
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"errors"
 	"net/url"
@@ -25,8 +26,6 @@ import (
 	"testing"
 	"time"
 
-	"context"
-
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/streamlog"
 	"vitess.io/vitess/go/vt/callinfo"
@@ -127,7 +126,7 @@ func TestLogStatsFormat(t *testing.T) {
 
 	*streamlog.QueryLogFormat = "text"
 	got = testFormat(logStats, url.Values(params))
-	want = "test\t\t\t''\t''\t2017-01-01 01:02:03.000000\t2017-01-01 01:02:04.000001\t1.000001\t\t\"sql\"\tmap[strVal:type:VARBINARY value:\"abc\"]\t1\t\"sql with pii\"\tmysql\t0.000000\t0.000000\t0\t12345\t1\t\"\"\t\n"
+	want = "test\t\t\t''\t''\t2017-01-01 01:02:03.000000\t2017-01-01 01:02:04.000001\t1.000001\t\t\"sql\"\tmap[strVal:type:VARCHAR value:\"abc\"]\t1\t\"sql with pii\"\tmysql\t0.000000\t0.000000\t0\t12345\t1\t\"\"\t\n"
 	if got != want {
 		t.Errorf("logstats format: got:\n%q\nwant:\n%q\n", got, want)
 	}
@@ -142,7 +141,7 @@ func TestLogStatsFormat(t *testing.T) {
 	if err != nil {
 		t.Errorf("logstats format: error marshaling json: %v -- got:\n%v", err, got)
 	}
-	want = "{\n    \"BindVars\": {\n        \"strVal\": {\n            \"type\": \"VARBINARY\",\n            \"value\": \"abc\"\n        }\n    },\n    \"CallInfo\": \"\",\n    \"ConnWaitTime\": 0,\n    \"Effective Caller\": \"\",\n    \"End\": \"2017-01-01 01:02:04.000001\",\n    \"Error\": \"\",\n    \"ImmediateCaller\": \"\",\n    \"Method\": \"test\",\n    \"MysqlTime\": 0,\n    \"OriginalSQL\": \"sql\",\n    \"PlanType\": \"\",\n    \"Queries\": 1,\n    \"QuerySources\": \"mysql\",\n    \"ResponseSize\": 1,\n    \"RewrittenSQL\": \"sql with pii\",\n    \"RowsAffected\": 0,\n    \"Start\": \"2017-01-01 01:02:03.000000\",\n    \"TotalTime\": 1.000001,\n    \"TransactionID\": 12345,\n    \"Username\": \"\"\n}"
+	want = "{\n    \"BindVars\": {\n        \"strVal\": {\n            \"type\": \"VARCHAR\",\n            \"value\": \"abc\"\n        }\n    },\n    \"CallInfo\": \"\",\n    \"ConnWaitTime\": 0,\n    \"Effective Caller\": \"\",\n    \"End\": \"2017-01-01 01:02:04.000001\",\n    \"Error\": \"\",\n    \"ImmediateCaller\": \"\",\n    \"Method\": \"test\",\n    \"MysqlTime\": 0,\n    \"OriginalSQL\": \"sql\",\n    \"PlanType\": \"\",\n    \"Queries\": 1,\n    \"QuerySources\": \"mysql\",\n    \"ResponseSize\": 1,\n    \"RewrittenSQL\": \"sql with pii\",\n    \"RowsAffected\": 0,\n    \"Start\": \"2017-01-01 01:02:03.000000\",\n    \"TotalTime\": 1.000001,\n    \"TransactionID\": 12345,\n    \"Username\": \"\"\n}"
 	if string(formatted) != want {
 		t.Errorf("logstats format: got:\n%q\nwant:\n%v\n", string(formatted), want)
 	}
diff --git a/go/vt/vttablet/tabletserver/tabletserver_test.go b/go/vt/vttablet/tabletserver/tabletserver_test.go
index e341146877..e31b85173c 100644
--- a/go/vt/vttablet/tabletserver/tabletserver_test.go
+++ b/go/vt/vttablet/tabletserver/tabletserver_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package tabletserver
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -38,8 +39,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"context"
-
 	"github.com/stretchr/testify/require"
 
 	"vitess.io/vitess/go/mysql"
@@ -1869,13 +1868,13 @@ func TestTruncateErrors(t *testing.T) {
 	)
 
 	// Error not truncated
-	wantErr := "sensitive message (errno 10) (sqlstate HY000): Sql: \"select * from test_table where xyz = :vtg1 order by abc desc\", BindVars: {vtg1: \"type:VARBINARY value:\\\"this is kinda long eh\\\"\"}"
+	wantErr := "sensitive message (errno 10) (sqlstate HY000): Sql: \"select * from test_table where xyz = :vtg1 order by abc desc\", BindVars: {vtg1: \"type:VARCHAR value:\\\"this is kinda long eh\\\"\"}"
 	if err == nil || err.Error() != wantErr {
 		t.Errorf("error got '%v', want '%s'", err, wantErr)
 	}
 
 	// but log *is* truncated
-	wantLog := "sensitive message (errno 10) (sqlstate HY000): Sql: \"select * from test_table where xyz = :vt [TRUNCATED]\", BindVars: {vtg1: \"type:VARBINARY value:\\ [TRUNCATED]"
+	wantLog := "sensitive message (errno 10) (sqlstate HY000): Sql: \"select * from test_table where xyz = :vt [TRUNCATED]\", BindVars: {vtg1: \"type:VARCHAR value:\\\"t [TRUNCATED]"
 	if wantLog != tl.getLog(0) {
 		t.Errorf("log got '%s', want '%s'", tl.getLog(0), wantLog)
 	}
@@ -1890,13 +1889,13 @@ func TestTruncateErrors(t *testing.T) {
 	)
 
 	// Error not truncated
-	wantErr = "sensitive message (errno 10) (sqlstate HY000): Sql: \"select * from test_table where xyz = :vtg1 order by abc desc\", BindVars: {vtg1: \"type:VARBINARY value:\\\"this is kinda long eh\\\"\"}"
+	wantErr = "sensitive message (errno 10) (sqlstate HY000): Sql: \"select * from test_table where xyz = :vtg1 order by abc desc\", BindVars: {vtg1: \"type:VARCHAR value:\\\"this is kinda long eh\\\"\"}"
 	if err == nil || err.Error() != wantErr {
 		t.Errorf("error got '%v', want '%s'", err, wantErr)
 	}
 
 	// Log not truncated, since our limit is large enough now
-	wantLog = "sensitive message (errno 10) (sqlstate HY000): Sql: \"select * from test_table where xyz = :vtg1 order by abc desc\", BindVars: {vtg1: \"type:VARBINARY value:\\\"this is kinda long eh\\\"\"}"
+	wantLog = "sensitive message (errno 10) (sqlstate HY000): Sql: \"select * from test_table where xyz = :vtg1 order by abc desc\", BindVars: {vtg1: \"type:VARCHAR value:\\\"this is kinda long eh\\\"\"}"
 	if wantLog != tl.getLog(1) {
 		t.Errorf("log got '%s', want '%s'", tl.getLog(1), wantLog)
 	}
diff --git a/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go b/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go
index 0a95523b3f..4ec57b15b7 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/planbuilder_test.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"testing"
 
+	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/vt/proto/topodata"
 
 	"github.com/stretchr/testify/require"
@@ -614,7 +615,7 @@ func TestPlanBuilderFilterComparison(t *testing.T) {
 	}, {
 		name:       "greater",
 		inFilter:   "select * from t1 where val > 'abc'",
-		outFilters: []Filter{{Opcode: GreaterThan, ColNum: 1, Value: sqltypes.NewVarBinary("abc")}},
+		outFilters: []Filter{{Opcode: GreaterThan, ColNum: 1, Value: sqltypes.NewVarChar("abc")}},
 	}, {
 		name:       "greater-than",
 		inFilter:   "select * from t1 where id >= 1",
@@ -623,7 +624,7 @@ func TestPlanBuilderFilterComparison(t *testing.T) {
 		name:     "less-than-with-and",
 		inFilter: "select * from t1 where id < 2 and val <= 'xyz'",
 		outFilters: []Filter{{Opcode: LessThan, ColNum: 0, Value: sqltypes.NewInt64(2)},
-			{Opcode: LessThanEqual, ColNum: 1, Value: sqltypes.NewVarBinary("xyz")},
+			{Opcode: LessThanEqual, ColNum: 1, Value: sqltypes.NewVarChar("xyz")},
 		},
 	}, {
 		name:     "vindex-and-operators",
@@ -641,7 +642,7 @@ func TestPlanBuilderFilterComparison(t *testing.T) {
 				},
 			},
 			{Opcode: Equal, ColNum: 0, Value: sqltypes.NewInt64(2)},
-			{Opcode: NotEqual, ColNum: 1, Value: sqltypes.NewVarBinary("xyz")},
+			{Opcode: NotEqual, ColNum: 1, Value: sqltypes.NewVarChar("xyz")},
 		},
 	}}
 
@@ -691,7 +692,7 @@ func TestCompare(t *testing.T) {
 	}
 	for _, tc := range testcases {
 		t.Run("", func(t *testing.T) {
-			got, err := compare(tc.opcode, tc.columnValue, tc.filterValue)
+			got, err := compare(tc.opcode, tc.columnValue, tc.filterValue, collations.CollationUtf8mb4ID)
 			require.NoError(t, err)
 			require.Equal(t, tc.want, got)
 		})