Repository: vitessio/vitess
Pull request #9017 (2021-10-19 06:42:12)

Gen4: Filter primitive and planning
## Description

This pull request adds the `Filter` engine primitive along with the `filter` logical plan. A basic version of the `ComparisonExpr` (implementation of EvalEngine's `Expr`) is added through this pull request too. `ComparisonExpr` supports basic operations (`=`, `!=`, `<`, `<=`, `>`, `>=`) on values that can be transformed to numerical values, i.e support for type coercion will come in the following pull request.

## Related Issue(s)

- #7280 

## Checklist
- [x] Tests were added or are not required
- [x] Documentation was added or is not required



Patch:
diff --git a/go/mysql/binlog_event_json.go b/go/mysql/binlog_event_json.go
index efc5648f32..7d76d67d32 100644
--- a/go/mysql/binlog_event_json.go
+++ b/go/mysql/binlog_event_json.go
@@ -23,8 +23,6 @@ import (
 
 	"vitess.io/vitess/go/vt/log"
 
-	"vitess.io/vitess/go/vt/vtgate/evalengine"
-
 	"github.com/spyzhov/ajson"
 
 	querypb "vitess.io/vitess/go/vt/proto/query"
@@ -474,7 +472,7 @@ func (oh opaquePlugin) getNode(typ jsonDataType, data []byte, pos int) (node *aj
 		if err != nil {
 			return nil, err
 		}
-		float, err := evalengine.ToFloat64(val)
+		float, err := val.ToFloat64()
 		if err != nil {
 			return nil, err
 		}
diff --git a/go/sqltypes/type.go b/go/sqltypes/type.go
index 45edc7b62b..350c8166b8 100644
--- a/go/sqltypes/type.go
+++ b/go/sqltypes/type.go
@@ -86,6 +86,11 @@ func IsNumber(t querypb.Type) bool {
 	return IsIntegral(t) || IsFloat(t) || t == Decimal
 }
 
+// IsDate returns true if the type represents a date and/or time.
+func IsDate(t querypb.Type) bool {
+	return t == Datetime || t == Date || t == Timestamp || t == Time
+}
+
 // Vitess data types. These are idiomatically
 // named synonyms for the querypb.Type values.
 // Although these constants are interchangeable,
diff --git a/go/sqltypes/value.go b/go/sqltypes/value.go
index 85002c8bc4..fd504460d3 100644
--- a/go/sqltypes/value.go
+++ b/go/sqltypes/value.go
@@ -143,6 +143,31 @@ func NewVarBinary(v string) Value {
 	return MakeTrusted(VarBinary, []byte(v))
 }
 
+// NewDate builds a Date value.
+func NewDate(v string) Value {
+	return MakeTrusted(Date, []byte(v))
+}
+
+// NewTime builds a Time value.
+func NewTime(v string) Value {
+	return MakeTrusted(Time, []byte(v))
+}
+
+// NewTimestamp builds a Timestamp value.
+func NewTimestamp(v string) Value {
+	return MakeTrusted(Timestamp, []byte(v))
+}
+
+// NewDatetime builds a Datetime value.
+func NewDatetime(v string) Value {
+	return MakeTrusted(Datetime, []byte(v))
+}
+
+// NewDecimal builds a Decimal value.
+func NewDecimal(v string) Value {
+	return MakeTrusted(Decimal, []byte(v))
+}
+
 // NewIntegral builds an integral type from a string representation.
 // The type will be Int64 or Uint64. Int64 will be preferred where possible.
 func NewIntegral(val string) (n Value, err error) {
diff --git a/go/vt/sqlparser/expression_converter.go b/go/vt/sqlparser/expression_converter.go
deleted file mode 100644
index e85fce2241..0000000000
--- a/go/vt/sqlparser/expression_converter.go
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
-Copyright 2020 The Vitess Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package sqlparser
-
-import (
-	"fmt"
-
-	"vitess.io/vitess/go/vt/vtgate/evalengine"
-)
-
-// ErrExprNotSupported signals that the expression cannot be handled by expression evaluation engine.
-var ErrExprNotSupported = fmt.Errorf("Expr Not Supported")
-
-//Convert converts between AST expressions and executable expressions
-func Convert(e Expr) (evalengine.Expr, error) {
-	switch node := e.(type) {
-	case Argument:
-		return evalengine.NewBindVar(string(node)), nil
-	case *Literal:
-		switch node.Type {
-		case IntVal:
-			return evalengine.NewLiteralIntFromBytes(node.Bytes())
-		case FloatVal:
-			return evalengine.NewLiteralFloat(node.Bytes())
-		case StrVal:
-			return evalengine.NewLiteralString(node.Bytes()), nil
-		}
-	case BoolVal:
-		if node {
-			return evalengine.NewLiteralIntFromBytes([]byte("1"))
-		}
-		return evalengine.NewLiteralIntFromBytes([]byte("0"))
-	case *BinaryExpr:
-		var op evalengine.BinaryExpr
-		switch node.Operator {
-		case PlusOp:
-			op = &evalengine.Addition{}
-		case MinusOp:
-			op = &evalengine.Subtraction{}
-		case MultOp:
-			op = &evalengine.Multiplication{}
-		case DivOp:
-			op = &evalengine.Division{}
-		default:
-			return nil, ErrExprNotSupported
-		}
-		left, err := Convert(node.Left)
-		if err != nil {
-			return nil, err
-		}
-		right, err := Convert(node.Right)
-		if err != nil {
-			return nil, err
-		}
-		return &evalengine.BinaryOp{
-			Expr:  op,
-			Left:  left,
-			Right: right,
-		}, nil
-
-	}
-	return nil, ErrExprNotSupported
-}
diff --git a/go/vt/vtgate/engine/cached_size.go b/go/vt/vtgate/engine/cached_size.go
index 46d00ace60..1af7b1076b 100644
--- a/go/vt/vtgate/engine/cached_size.go
+++ b/go/vt/vtgate/engine/cached_size.go
@@ -175,6 +175,28 @@ func (cached *Distinct) CachedSize(alloc bool) int64 {
 	}
 	return size
 }
+func (cached *Filter) CachedSize(alloc bool) int64 {
+	if cached == nil {
+		return int64(0)
+	}
+	size := int64(0)
+	if alloc {
+		size += int64(48)
+	}
+	// field Predicate vitess.io/vitess/go/vt/vtgate/evalengine.Expr
+	if cc, ok := cached.Predicate.(cachedObject); ok {
+		size += cc.CachedSize(true)
+	}
+	// field ASTPredicate vitess.io/vitess/go/vt/sqlparser.Expr
+	if cc, ok := cached.ASTPredicate.(cachedObject); ok {
+		size += cc.CachedSize(true)
+	}
+	// field Input vitess.io/vitess/go/vt/vtgate/engine.Primitive
+	if cc, ok := cached.Input.(cachedObject); ok {
+		size += cc.CachedSize(true)
+	}
+	return size
+}
 func (cached *Gen4CompareV3) CachedSize(alloc bool) int64 {
 	if cached == nil {
 		return int64(0)
@@ -702,7 +724,7 @@ func (cached *SemiJoin) CachedSize(alloc bool) int64 {
 	}
 	size := int64(0)
 	if alloc {
-		size += int64(48)
+		size += int64(64)
 	}
 	// field Left vitess.io/vitess/go/vt/vtgate/engine.Primitive
 	if cc, ok := cached.Left.(cachedObject); ok {
@@ -712,6 +734,10 @@ func (cached *SemiJoin) CachedSize(alloc bool) int64 {
 	if cc, ok := cached.Right.(cachedObject); ok {
 		size += cc.CachedSize(true)
 	}
+	// field Cols []int
+	{
+		size += hack.RuntimeAllocSize(int64(cap(cached.Cols)) * int64(8))
+	}
 	// field Vars map[string]int
 	if cached.Vars != nil {
 		size += int64(48)
diff --git a/go/vt/vtgate/engine/filter.go b/go/vt/vtgate/engine/filter.go
new file mode 100644
index 0000000000..27480e17c4
--- /dev/null
+++ b/go/vt/vtgate/engine/filter.go
@@ -0,0 +1,128 @@
+/*
+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 engine
+
+import (
+	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/vtgate/evalengine"
+
+	"vitess.io/vitess/go/sqltypes"
+
+	querypb "vitess.io/vitess/go/vt/proto/query"
+)
+
+var _ Primitive = (*Filter)(nil)
+
+// Filter is a primitive that performs the FILTER operation.
+type Filter struct {
+	Predicate    evalengine.Expr
+	ASTPredicate sqlparser.Expr
+	Input        Primitive
+
+	noTxNeeded
+}
+
+// RouteType returns a description of the query routing type used by the primitive
+func (f *Filter) RouteType() string {
+	return f.Input.RouteType()
+}
+
+// GetKeyspaceName specifies the Keyspace that this primitive routes to.
+func (f *Filter) GetKeyspaceName() string {
+	return f.Input.GetKeyspaceName()
+}
+
+// GetTableName specifies the table that this primitive routes to.
+func (f *Filter) GetTableName() string {
+	return f.Input.GetTableName()
+}
+
+// TryExecute satisfies the Primitive interface.
+func (f *Filter) TryExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool) (*sqltypes.Result, error) {
+	result, err := f.Input.TryExecute(vcursor, bindVars, wantfields)
+	if err != nil {
+		return nil, err
+	}
+	env := evalengine.ExpressionEnv{
+		BindVars: bindVars,
+	}
+	var rows [][]sqltypes.Value
+	for _, row := range result.Rows {
+		env.Row = row
+		evalResult, err := f.Predicate.Evaluate(env)
+		if err != nil {
+			return nil, err
+		}
+		intEvalResult, err := evalResult.Value().ToInt64()
+		if err != nil {
+			return nil, err
+		}
+		if intEvalResult == 1 {
+			rows = append(rows, row)
+		}
+	}
+	result.Rows = rows
+	return result, nil
+}
+
+// TryStreamExecute satisfies the Primitive interface.
+func (f *Filter) TryStreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
+	env := evalengine.ExpressionEnv{
+		BindVars: bindVars,
+	}
+	filter := func(results *sqltypes.Result) error {
+		var rows [][]sqltypes.Value
+		for _, row := range results.Rows {
+			env.Row = row
+			evalResult, err := f.Predicate.Evaluate(env)
+			if err != nil {
+				return err
+			}
+			intEvalResult, err := evalResult.Value().ToInt64()
+			if err != nil {
+				return err
+			}
+			if intEvalResult == 1 {
+				rows = append(rows, row)
+			}
+		}
+		results.Rows = rows
+		return callback(results)
+	}
+	return f.Input.TryStreamExecute(vcursor, bindVars, wantfields, filter)
+}
+
+// GetFields implements the Primitive interface.
+func (f *Filter) GetFields(vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
+	return f.Input.GetFields(vcursor, bindVars)
+}
+
+// Inputs returns the input to limit
+func (f *Filter) Inputs() []Primitive {
+	return []Primitive{f.Input}
+}
+
+func (f *Filter) description() PrimitiveDescription {
+	other := map[string]interface{}{
+		"Predicate": sqlparser.String(f.ASTPredicate),
+	}
+
+	return PrimitiveDescription{
+		OperatorType: "Filter",
+		Other:        other,
+	}
+}
diff --git a/go/vt/vtgate/engine/ordered_aggregate.go b/go/vt/vtgate/engine/ordered_aggregate.go
index c8125577e2..be0d5f9bce 100644
--- a/go/vt/vtgate/engine/ordered_aggregate.go
+++ b/go/vt/vtgate/engine/ordered_aggregate.go
@@ -449,15 +449,15 @@ func (oa *OrderedAggregate) merge(fields []*querypb.Field, row1, row2 []sqltypes
 		case AggregateCount, AggregateSum:
 			value := row1[aggr.Col]
 			v2 := row2[aggr.Col]
-			result[aggr.Col] = evalengine.NullsafeAdd(value, v2, fields[aggr.Col].Type)
+			result[aggr.Col], err = evalengine.NullSafeAdd(value, v2, fields[aggr.Col].Type)
 		case AggregateMin:
 			result[aggr.Col], err = evalengine.Min(row1[aggr.Col], row2[aggr.Col], colls[aggr.Col])
 		case AggregateMax:
 			result[aggr.Col], err = evalengine.Max(row1[aggr.Col], row2[aggr.Col], colls[aggr.Col])
 		case AggregateCountDistinct:
-			result[aggr.Col] = evalengine.NullsafeAdd(row1[aggr.Col], countOne, OpcodeType[aggr.Opcode])
+			result[aggr.Col], err = evalengine.NullSafeAdd(row1[aggr.Col], countOne, OpcodeType[aggr.Opcode])
 		case AggregateSumDistinct:
-			result[aggr.Col] = evalengine.NullsafeAdd(row1[aggr.Col], row2[aggr.Col], OpcodeType[aggr.Opcode])
+			result[aggr.Col], err = evalengine.NullSafeAdd(row1[aggr.Col], row2[aggr.Col], OpcodeType[aggr.Opcode])
 		case AggregateGtid:
 			vgtid := &binlogdatapb.VGtid{}
 			err = proto.Unmarshal(row1[aggr.Col].ToBytes(), vgtid)
diff --git a/go/vt/vtgate/evalengine/arithmetic.go b/go/vt/vtgate/evalengine/arithmetic.go
index c96b5d4950..90e6379d96 100644
--- a/go/vt/vtgate/evalengine/arithmetic.go
+++ b/go/vt/vtgate/evalengine/arithmetic.go
@@ -56,6 +56,10 @@ func (err UnsupportedCollationError) Error() string {
 	return fmt.Sprintf("comparison using collation %d isn't possible", err.ID)
 }
 
+func dataOutOfRangeError(v1, v2 interface{}, typ, sign string) error {
+	return vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.DataOutOfRange, "%s value is out of range in %v %s %v", typ, v1, sign, v2)
+}
+
 // Add adds two values together
 // if v1 or v2 is null, then it returns null
 func Add(v1, v2 sqltypes.Value) (sqltypes.Value, error) {
@@ -158,7 +162,7 @@ func Divide(v1, v2 sqltypes.Value) (sqltypes.Value, error) {
 	return lresult.toSQLValue(lresult.typ), nil
 }
 
-// NullsafeAdd adds two Values in a null-safe manner. A null value
+// NullSafeAdd adds two Values in a null-safe manner. A null value
 // is treated as 0. If both values are null, then a null is returned.
 // If both values are not null, a numeric value is built
 // from each input: Signed->int64, Unsigned->uint64, Float->float64.
@@ -169,7 +173,7 @@ func Divide(v1, v2 sqltypes.Value) (sqltypes.Value, error) {
 // addition, if one of the input types was Decimal, then
 // a Decimal is built. Otherwise, the final type of the
 // result is preserved.
-func NullsafeAdd(v1, v2 sqltypes.Value, resultType querypb.Type) sqltypes.Value {
+func NullSafeAdd(v1, v2 sqltypes.Value, resultType querypb.Type) (sqltypes.Value, error) {
 	if v1.IsNull() {
 		v1 = sqltypes.MakeTrusted(resultType, zeroBytes)
 	}
@@ -179,15 +183,17 @@ func NullsafeAdd(v1, v2 sqltypes.Value, resultType querypb.Type) sqltypes.Value
 
 	lv1, err := newEvalResult(v1)
 	if err != nil {
-		return sqltypes.NULL
+		return sqltypes.NULL, err
 	}
 	lv2, err := newEvalResult(v2)
 	if err != nil {
-		return sqltypes.NULL
+		return sqltypes.NULL, err
 	}
-	lresult := addNumeric(lv1, lv2)
-
-	return lresult.toSQLValue(resultType)
+	lresult, err := addNumericWithError(lv1, lv2)
+	if err != nil {
+		return sqltypes.NULL, err
+	}
+	return lresult.toSQLValue(resultType), nil
 }
 
 // NullsafeCompare returns 0 if v1==v2, -1 if v1<v2, and 1 if v1>v2.
@@ -224,7 +230,7 @@ func NullsafeCompare(v1, v2 sqltypes.Value, collationID collations.ID) (int, err
 		return bytes.Compare(v1.ToBytes(), v2.ToBytes()), nil
 	}
 	if v1.IsText() && v2.IsText() && collationID != collations.Unknown {
-		collation := collations.FromID(collationID)
+		collation := collations.Default().LookupByID(collationID)
 		if collation == nil {
 			return 0, UnsupportedCollationError{
 				ID: collationID,
@@ -312,7 +318,7 @@ func minmax(v1, v2 sqltypes.Value, min bool, collation collations.ID) (sqltypes.
 }
 
 func addNumeric(v1, v2 EvalResult) EvalResult {
-	v1, v2 = makeNumericAndprioritize(v1, v2)
+	v1, v2 = makeNumericAndPrioritize(v1, v2)
 	switch v1.typ {
 	case sqltypes.Int64:
 		return intPlusInt(v1.ival, v2.ival)
@@ -330,7 +336,7 @@ func addNumeric(v1, v2 EvalResult) EvalResult {
 }
 
 func addNumericWithError(v1, v2 EvalResult) (EvalResult, error) {
-	v1, v2 = makeNumericAndprioritize(v1, v2)
+	v1, v2 = makeNumericAndPrioritize(v1, v2)
 	switch v1.typ {
 	case sqltypes.Int64:
 		return intPlusIntWithError(v1.ival, v2.ival)
@@ -341,7 +347,7 @@ func addNumericWithError(v1, v2 EvalResult) (EvalResult, error) {
 		case sqltypes.Uint64:
 			return uintPlusUintWithError(v1.uval, v2.uval)
 		}
-	case sqltypes.Float64:
+	case sqltypes.Float64, sqltypes.Decimal:
 		return floatPlusAny(v1.fval, v2), nil
 	}
 	return EvalResult{}, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "invalid arithmetic between: %s %s", v1.Value().String(), v2.Value().String())
@@ -377,7 +383,7 @@ func subtractNumericWithError(i1, i2 EvalResult) (EvalResult, error) {
 }
 
 func multiplyNumericWithError(v1, v2 EvalResult) (EvalResult, error) {
-	v1, v2 = makeNumericAndprioritize(v1, v2)
+	v1, v2 = makeNumericAndPrioritize(v1, v2)
 	switch v1.typ {
 	case sqltypes.Int64:
 		return intTimesIntWithError(v1.ival, v2.ival)
@@ -411,9 +417,9 @@ func divideNumericWithError(i1, i2 EvalResult) (EvalResult, error) {
 	return EvalResult{}, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "invalid arithmetic between: %s %s", v1.Value().String(), v2.Value().String())
 }
 
-// makeNumericAndprioritize reorders the input parameters
+// makeNumericAndPrioritize reorders the input parameters
 // to be Float64, Uint64, Int64.
-func makeNumericAndprioritize(i1, i2 EvalResult) (EvalResult, EvalResult) {
+func makeNumericAndPrioritize(i1, i2 EvalResult) (EvalResult, EvalResult) {
 	v1 := makeNumeric(i1)
 	v2 := makeNumeric(i2)
 	switch v1.typ {
@@ -429,6 +435,19 @@ func makeNumericAndprioritize(i1, i2 EvalResult) (EvalResult, EvalResult) {
 	return v1, v2
 }
 
+func makeFloat(v EvalResult) EvalResult {
+	if sqltypes.IsIntegral(v.typ) {
+		return EvalResult{fval: float64(v.ival), typ: sqltypes.Float64}
+	}
+	if sqltypes.IsFloat(v.typ) {
+		return v
+	}
+	if fval, err := strconv.ParseFloat(string(v.bytes), 64); err == nil {
+		return EvalResult{fval: fval, typ: sqltypes.Float64}
+	}
+	return EvalResult{ival: 0, typ: sqltypes.Int64}
+}
+
 func makeNumeric(v EvalResult) EvalResult {
 	if sqltypes.IsNumber(v.typ) {
 		return v
@@ -459,7 +478,7 @@ overflow:
 func intPlusIntWithError(v1, v2 int64) (EvalResult, error) {
 	result := v1 + v2
 	if (result > v1) != (v2 > 0) {
-		return EvalResult{}, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.DataOutOfRange, "%s value is out of range in %v + %v", "BIGINT", v1, v2)
+		return EvalResult{}, dataOutOfRangeError(v1, v2, "BIGINT", "+")
 	}
 	return EvalResult{typ: sqltypes.Int64, ival: result}, nil
 }
@@ -468,7 +487,7 @@ func intMinusIntWithError(v1, v2 int64) (EvalResult, error) {
 	result := v1 - v2
 
 	if (result < v1) != (v2 > 0) {
-		return EvalResult{}, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.DataOutOfRange, "%s value is out of range in %v - %v", "BIGINT", v1, v2)
+		return EvalResult{}, dataOutOfRangeError(v1, v2, "BIGINT", "-")
 	}
 	return EvalResult{typ: sqltypes.Int64, ival: result}, nil
 }
@@ -476,7 +495,7 @@ func intMinusIntWithError(v1, v2 int64) (EvalResult, error) {
 func intTimesIntWithError(v1, v2 int64) (EvalResult, error) {
 	result := v1 * v2
 	if v1 != 0 && result/v1 != v2 {
-		return EvalResult{}, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.DataOutOfRange, "%s value is out of range in %v * %v", "BIGINT", v1, v2)
+		return EvalResult{}, dataOutOfRangeError(v1, v2, "BIGINT", "*")
 	}
 	return EvalResult{typ: sqltypes.Int64, ival: result}, nil
 
@@ -484,7 +503,7 @@ func intTimesIntWithError(v1, v2 int64) (EvalResult, error) {
 
 func intMinusUintWithError(v1 int64, v2 uint64) (EvalResult, error) {
 	if v1 < 0 || v1 < int64(v2) {
-		return EvalResult{}, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.DataOutOfRange, "%s value is out of range in %v - %v", "BIGINT UNSIGNED", v1, v2)
+		return EvalResult{}, dataOutOfRangeError(v1, v2, "BIGINT UNSIGNED", "-")
 	}
 	return uintMinusUintWithError(uint64(v1), v2)
 }
@@ -494,17 +513,18 @@ func uintPlusInt(v1 uint64, v2 int64) EvalResult {
 }
 
 func uintPlusIntWithError(v1 uint64, v2 int64) (EvalResult, error) {
-	if v2 < 0 && v1 < uint64(v2) {
-		return EvalResult{}, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.DataOutOfRange, "%s value is out of range in %v + %v", "BIGINT UNSIGNED", v1, v2)
+	result := v1 + uint64(v2)
+	if v2 < 0 && v1 < uint64(-v2) || v2 > 0 && (result < v1 || result < uint64(v2)) {
+		return EvalResult{}, dataOutOfRangeError(v1, v2, "BIGINT UNSIGNED", "+")
 	}
 	// convert to int -> uint is because for numeric operators (such as + or -)
 	// where one of the operands is an unsigned integer, the result is unsigned by default.
-	return uintPlusUintWithError(v1, uint64(v2))
+	return EvalResult{typ: sqltypes.Uint64, uval: result}, nil
 }
 
 func uintMinusIntWithError(v1 uint64, v2 int64) (EvalResult, error) {
 	if int64(v1) < v2 && v2 > 0 {
-		return EvalResult{}, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.DataOutOfRange, "%s value is out of range in %v - %v", "BIGINT UNSIGNED", v1, v2)
+		return EvalResult{}, dataOutOfRangeError(v1, v2, "BIGINT UNSIGNED", "-")
 	}
 	// uint - (- int) = uint + int
 	if v2 < 0 {
@@ -515,7 +535,7 @@ func uintMinusIntWithError(v1 uint64, v2 int64) (EvalResult, error) {
 
 func uintTimesIntWithError(v1 uint64, v2 int64) (EvalResult, error) {
 	if v2 < 0 || int64(v1) < 0 {
-		return EvalResult{}, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.DataOutOfRange, "%s value is out of range in %v * %v", "BIGINT UNSIGNED", v1, v2)
+		return EvalResult{}, dataOutOfRangeError(v1, v2, "BIGINT UNSIGNED", "*")
 	}
 	return uintTimesUintWithError(v1, uint64(v2))
 }
@@ -530,8 +550,8 @@ func uintPlusUint(v1, v2 uint64) EvalResult {
 
 func uintPlusUintWithError(v1, v2 uint64) (EvalResult, error) {
 	result := v1 + v2
-	if result < v2 {
-		return EvalResult{}, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.DataOutOfRange, "%s value is out of range in %v + %v", "BIGINT UNSIGNED", v1, v2)
+	if result < v1 || result < v2 {
+		return EvalResult{}, dataOutOfRangeError(v1, v2, "BIGINT UNSIGNED", "+")
 	}
 	return EvalResult{typ: sqltypes.Uint64, uval: result}, nil
 }
@@ -539,7 +559,7 @@ func uintPlusUintWithError(v1, v2 uint64) (EvalResult, error) {
 func uintMinusUintWithError(v1, v2 uint64) (EvalResult, error) {
 	result := v1 - v2
 	if v2 > v1 {
-		return EvalResult{}, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.DataOutOfRange, "%s value is out of range in %v - %v", "BIGINT UNSIGNED", v1, v2)
+		return EvalResult{}, dataOutOfRangeError(v1, v2, "BIGINT UNSIGNED", "-")
 	}
 
 	return EvalResult{typ: sqltypes.Uint64, uval: result}, nil
@@ -548,7 +568,7 @@ func uintMinusUintWithError(v1, v2 uint64) (EvalResult, error) {
 func uintTimesUintWithError(v1, v2 uint64) (EvalResult, error) {
 	result := v1 * v2
 	if result < v2 || result < v1 {
-		return EvalResult{}, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.DataOutOfRange, "%s value is out of range in %v * %v", "BIGINT UNSIGNED", v1, v2)
+		return EvalResult{}, dataOutOfRangeError(v1, v2, "BIGINT UNSIGNED", "*")
 	}
 	return EvalResult{typ: sqltypes.Uint64, uval: result}, nil
 }
diff --git a/go/vt/vtgate/evalengine/binary.go b/go/vt/vtgate/evalengine/binary.go
new file mode 100644
index 0000000000..8637f7b302
--- /dev/null
+++ b/go/vt/vtgate/evalengine/binary.go
@@ -0,0 +1,139 @@
+/*
+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 (
+	"vitess.io/vitess/go/sqltypes"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+)
+
+type (
+	// BinaryOp allows binary expressions to not have to evaluate child expressions - this is done by the BinaryExpr
+	BinaryOp interface {
+		Evaluate(left, right EvalResult) (EvalResult, error)
+		Type(left querypb.Type) querypb.Type
+		String() string
+	}
+
+	BinaryExpr struct {
+		Op          BinaryOp
+		Left, Right Expr
+	}
+
+	// Binary ops
+	Addition       struct{}
+	Subtraction    struct{}
+	Multiplication struct{}
+	Division       struct{}
+)
+
+var _ BinaryOp = (*Addition)(nil)
+var _ BinaryOp = (*Subtraction)(nil)
+var _ BinaryOp = (*Multiplication)(nil)
+var _ BinaryOp = (*Division)(nil)
+
+// Evaluate implements the Expr interface
+func (b *BinaryExpr) Evaluate(env ExpressionEnv) (EvalResult, error) {
+	lVal, err := b.Left.Evaluate(env)
+	if err != nil {
+		return EvalResult{}, err
+	}
+	rVal, err := b.Right.Evaluate(env)
+	if err != nil {
+		return EvalResult{}, err
+	}
+	return b.Op.Evaluate(lVal, rVal)
+}
+
+// Type implements the Expr interface
+func (b *BinaryExpr) Type(env ExpressionEnv) (querypb.Type, error) {
+	ltype, err := b.Left.Type(env)
+	if err != nil {
+		return 0, err
+	}
+	rtype, err := b.Right.Type(env)
+	if err != nil {
+		return 0, err
+	}
+	typ := mergeNumericalTypes(ltype, rtype)
+	return b.Op.Type(typ), nil
+}
+
+// String implements the Expr interface
+func (b *BinaryExpr) String() string {
+	return b.Left.String() + " " + b.Op.String() + " " + b.Right.String()
+}
+
+// Evaluate implements the BinaryOp interface
+func (a *Addition) Evaluate(left, right EvalResult) (EvalResult, error) {
+	return addNumericWithError(left, right)
+}
+
+// Evaluate implements the BinaryOp interface
+func (s *Subtraction) Evaluate(left, right EvalResult) (EvalResult, error) {
+	return subtractNumericWithError(left, right)
+}
+
+// Evaluate implements the BinaryOp interface
+func (m *Multiplication) Evaluate(left, right EvalResult) (EvalResult, error) {
+	return multiplyNumericWithError(left, right)
+}
+
+// Evaluate implements the BinaryOp interface
+func (d *Division) Evaluate(left, right EvalResult) (EvalResult, error) {
+	return divideNumericWithError(left, right)
+}
+
+// Type implements the BinaryOp interface
+func (a *Addition) Type(left querypb.Type) querypb.Type {
+	return left
+}
+
+// Type implements the BinaryOp interface
+func (s *Subtraction) Type(left querypb.Type) querypb.Type {
+	return left
+}
+
+// Type implements the BinaryOp interface
+func (m *Multiplication) Type(left querypb.Type) querypb.Type {
+	return left
+}
+
+// Type implements the BinaryOp interface
+func (d *Division) Type(querypb.Type) querypb.Type {
+	return sqltypes.Float64
+}
+
+// String implements the BinaryOp interface
+func (a *Addition) String() string {
+	return "+"
+}
+
+// String implements the BinaryOp interface
+func (s *Subtraction) String() string {
+	return "-"
+}
+
+// String implements the BinaryOp interface
+func (m *Multiplication) String() string {
+	return "*"
+}
+
+// String implements the BinaryOp interface
+func (d *Division) String() string {
+	return "/"
+}
diff --git a/go/vt/vtgate/evalengine/cached_size.go b/go/vt/vtgate/evalengine/cached_size.go
index daee0218b6..dced4b9e4d 100644
--- a/go/vt/vtgate/evalengine/cached_size.go
+++ b/go/vt/vtgate/evalengine/cached_size.go
@@ -23,7 +23,7 @@ type cachedObject interface {
 	CachedSize(alloc bool) int64
 }
 
-func (cached *BinaryOp) CachedSize(alloc bool) int64 {
+func (cached *BinaryExpr) CachedSize(alloc bool) int64 {
 	if cached == nil {
 		return int64(0)
 	}
@@ -31,8 +31,8 @@ func (cached *BinaryOp) CachedSize(alloc bool) int64 {
 	if alloc {
 		size += int64(48)
 	}
-	// field Expr vitess.io/vitess/go/vt/vtgate/evalengine.BinaryExpr
-	if cc, ok := cached.Expr.(cachedObject); ok {
+	// field Op vitess.io/vitess/go/vt/vtgate/evalengine.BinaryOp
+	if cc, ok := cached.Op.(cachedObject); ok {
 		size += cc.CachedSize(true)
 	}
 	// field Left vitess.io/vitess/go/vt/vtgate/evalengine.Expr
@@ -51,7 +51,7 @@ func (cached *BindVariable) CachedSize(alloc bool) int64 {
 	}
 	size := int64(0)
 	if alloc {
-		size += int64(16)
+		size += int64(24)
 	}
 	// field Key string
 	size += hack.RuntimeAllocSize(int64(len(cached.Key)))
@@ -63,7 +63,29 @@ func (cached *Column) CachedSize(alloc bool) int64 {
 	}
 	size := int64(0)
 	if alloc {
-		size += int64(8)
+		size += int64(16)
+	}
+	return size
+}
+func (cached *ComparisonExpr) CachedSize(alloc bool) int64 {
+	if cached == nil {
+		return int64(0)
+	}
+	size := int64(0)
+	if alloc {
+		size += int64(48)
+	}
+	// field Op vitess.io/vitess/go/vt/vtgate/evalengine.ComparisonOp
+	if cc, ok := cached.Op.(cachedObject); ok {
+		size += cc.CachedSize(true)
+	}
+	// field Left vitess.io/vitess/go/vt/vtgate/evalengine.Expr
+	if cc, ok := cached.Left.(cachedObject); ok {
+		size += cc.CachedSize(true)
+	}
+	// field Right vitess.io/vitess/go/vt/vtgate/evalengine.Expr
+	if cc, ok := cached.Right.(cachedObject); ok {
+		size += cc.CachedSize(true)
 	}
 	return size
 }
@@ -85,7 +107,7 @@ func (cached *Literal) CachedSize(alloc bool) int64 {
 	}
 	size := int64(0)
 	if alloc {
-		size += int64(64)
+		size += int64(80)
 	}
 	// field Val vitess.io/vitess/go/vt/vtgate/evalengine.EvalResult
 	size += cached.Val.CachedSize(false)
diff --git a/go/vt/vtgate/evalengine/comparisons.go b/go/vt/vtgate/evalengine/comparisons.go
new file mode 100644
index 0000000000..22350605ef
--- /dev/null
+++ b/go/vt/vtgate/evalengine/comparisons.go
@@ -0,0 +1,494 @@
+/*
+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 (
+	"vitess.io/vitess/go/sqltypes"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/vterrors"
+)
+
+type (
+	// ComparisonOp interfaces all the possible comparison operations we have, it eases the job of ComparisonExpr
+	// when evaluating the whole comparison
+	ComparisonOp interface {
+		Evaluate(left, right EvalResult) (EvalResult, error)
+		IsTrue(left, right EvalResult) (bool, error)
+		Type() querypb.Type
+		String() string
+	}
+
+	ComparisonExpr struct {
+		Op          ComparisonOp
+		Left, Right Expr
+	}
+
+	EqualOp         struct{}
+	NotEqualOp      struct{}
+	NullSafeEqualOp struct{}
+	LessThanOp      struct{}
+	LessEqualOp     struct{}
+	GreaterThanOp   struct{}
+	GreaterEqualOp  struct{}
+	InOp            struct{}
+	NotInOp         struct{}
+	LikeOp          struct{}
+	NotLikeOp       struct{}
+	RegexpOp        struct{}
+	NotRegexpOp     struct{}
+)
+
+var (
+	resultTrue  = EvalResult{typ: sqltypes.Int32, ival: 1}
+	resultFalse = EvalResult{typ: sqltypes.Int32, ival: 0}
+	resultNull  = EvalResult{typ: sqltypes.Null}
+)
+
+var _ ComparisonOp = (*EqualOp)(nil)
+var _ ComparisonOp = (*NotEqualOp)(nil)
+var _ ComparisonOp = (*NullSafeEqualOp)(nil)
+var _ ComparisonOp = (*LessThanOp)(nil)
+var _ ComparisonOp = (*LessEqualOp)(nil)
+var _ ComparisonOp = (*GreaterThanOp)(nil)
+var _ ComparisonOp = (*GreaterEqualOp)(nil)
+var _ ComparisonOp = (*InOp)(nil)
+var _ ComparisonOp = (*NotInOp)(nil)
+var _ ComparisonOp = (*LikeOp)(nil)
+var _ ComparisonOp = (*NotLikeOp)(nil)
+var _ ComparisonOp = (*RegexpOp)(nil)
+var _ ComparisonOp = (*NotRegexpOp)(nil)
+
+func (c *ComparisonExpr) evaluateComparisonExprs(env ExpressionEnv) (EvalResult, EvalResult, error) {
+	var lVal, rVal EvalResult
+	var err error
+	if lVal, err = c.Left.Evaluate(env); err != nil {
+		return EvalResult{}, EvalResult{}, err
+	}
+	if rVal, err = c.Right.Evaluate(env); err != nil {
+		return EvalResult{}, EvalResult{}, err
+	}
+	return lVal, rVal, nil
+}
+
+func hasNullEvalResult(l, r EvalResult) bool {
+	return l.typ == sqltypes.Null || r.typ == sqltypes.Null
+}
+
+func evalResultsAreStrings(l, r EvalResult) bool {
+	return (sqltypes.IsText(l.typ) || sqltypes.IsBinary(l.typ)) && (sqltypes.IsText(r.typ) || sqltypes.IsBinary(r.typ))
+}
+
+func evalResultsAreSameNumericType(l, r EvalResult) bool {
+	if sqltypes.IsIntegral(l.typ) && sqltypes.IsIntegral(r.typ) {
+		return true
+	}
+	if sqltypes.IsFloat(l.typ) && sqltypes.IsFloat(r.typ) {
+		return true
+	}
+	if l.typ == sqltypes.Decimal && r.typ == sqltypes.Decimal {
+		return true
+	}
+	return false
+}
+
+func needsDecimalHandling(l, r EvalResult) bool {
+	return l.typ == sqltypes.Decimal && (sqltypes.IsIntegral(r.typ) || sqltypes.IsFloat(r.typ)) ||
+		r.typ == sqltypes.Decimal && (sqltypes.IsIntegral(l.typ) || sqltypes.IsFloat(l.typ))
+}
+
+func evalResultsAreDates(l, r EvalResult) bool {
+	return sqltypes.IsDate(l.typ) && sqltypes.IsDate(r.typ)
+}
+
+func evalResultsAreDateAndString(l, r EvalResult) bool {
+	return sqltypes.IsDate(l.typ) && (sqltypes.IsText(r.typ) || sqltypes.IsBinary(r.typ)) ||
+		(sqltypes.IsText(l.typ) || sqltypes.IsBinary(l.typ)) && sqltypes.IsDate(r.typ)
+}
+
+func evalResultsAreDateAndNumeric(l, r EvalResult) bool {
+	return sqltypes.IsDate(l.typ) && sqltypes.IsNumber(r.typ) || sqltypes.IsNumber(l.typ) && sqltypes.IsDate(r.typ)
+}
+
+// For more details on comparison expression evaluation and type conversion:
+// 		- https://dev.mysql.com/doc/refman/8.0/en/type-conversion.html
+func executeComparison(lVal, rVal EvalResult) (int, error) {
+	switch {
+	case evalResultsAreStrings(lVal, rVal):
+		return compareStrings(lVal, rVal)
+
+	case evalResultsAreSameNumericType(lVal, rVal), needsDecimalHandling(lVal, rVal):
+		return compareNumeric(lVal, rVal)
+
+	case evalResultsAreDates(lVal, rVal):
+		return compareDates(lVal, rVal)
+
+	case evalResultsAreDateAndString(lVal, rVal):
+		return compareDateAndString(lVal, rVal)
+
+	case evalResultsAreDateAndNumeric(lVal, rVal):
+		// TODO: support comparison between a date and a numeric value
+		// 		queries like the ones below should be supported:
+		// 			- select 1 where 20210101 = cast("2021-01-01" as date)
+		// 			- select 1 where 2021210101 = cast("2021-01-01" as date)
+		// 			- select 1 where 104200 = cast("10:42:00" as time)
+		return 0, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "cannot compare a date with a numeric value")
+
+	default:
+		// Quoting MySQL Docs:
+		//
+		// 		"In all other cases, the arguments are compared as floating-point (real) numbers.
+		// 		For example, a comparison of string and numeric operands takes place as a
+		// 		comparison of floating-point numbers."
+		//
+		//		https://dev.mysql.com/doc/refman/8.0/en/type-conversion.html
+		return compareNumeric(makeFloat(lVal), makeFloat(rVal))
+	}
+}
+
+// Evaluate implements the Expr interface
+func (c *ComparisonExpr) Evaluate(env ExpressionEnv) (EvalResult, error) {
+	if c.Op == nil {
+		return EvalResult{}, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "a comparison expression needs a comparison operator")
+	}
+
+	lVal, rVal, err := c.evaluateComparisonExprs(env)
+	if err != nil {
+		return EvalResult{}, err
+	}
+
+	if hasNullEvalResult(lVal, rVal) {
+		// Comparison operation NullSafeEqual (<=>) does not care if one or two sides are NULL
+		if _, isNullsafe := c.Op.(*NullSafeEqualOp); !isNullsafe {
+			// If a side of the comparison is NULL, result will always be NULL
+			return resultNull, nil
+		}
+	}
+	return c.Op.Evaluate(lVal, rVal)
+}
+
+// Type implements the Expr interface
+func (c *ComparisonExpr) Type(ExpressionEnv) (querypb.Type, error) {
+	return querypb.Type_INT32, nil
+}
+
+// String implements the Expr interface
+func (c *ComparisonExpr) String() string {
+	return c.Left.String() + " " + c.Op.String() + " " + c.Right.String()
+}
+
+// Evaluate implements the ComparisonOp interface
+func (e *EqualOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	if out, err := e.IsTrue(left, right); err != nil || !out {
+		return resultFalse, err
+	}
+	return resultTrue, nil
+}
+
+// IsTrue implements the ComparisonOp interface
+func (e *EqualOp) IsTrue(left, right EvalResult) (bool, error) {
+	numeric, err := executeComparison(left, right)
+	if err != nil {
+		return false, err
+	}
+	return numeric == 0, nil
+}
+
+// Type implements the ComparisonOp interface
+func (e *EqualOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (e *EqualOp) String() string {
+	return "="
+}
+
+// Evaluate implements the ComparisonOp interface
+func (n *NotEqualOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	if out, err := n.IsTrue(left, right); err != nil || !out {
+		return resultFalse, err
+	}
+	return resultTrue, nil
+}
+
+// IsTrue implements the ComparisonOp interface
+func (n *NotEqualOp) IsTrue(left, right EvalResult) (bool, error) {
+	numeric, err := executeComparison(left, right)
+	if err != nil {
+		return false, err
+	}
+	return numeric != 0, nil
+}
+
+// Type implements the ComparisonOp interface
+func (n *NotEqualOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (n *NotEqualOp) String() string {
+	return "!="
+}
+
+// Evaluate implements the ComparisonOp interface
+func (n *NullSafeEqualOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	panic("implement me")
+}
+
+// IsTrue implements the ComparisonOp interface
+func (n *NullSafeEqualOp) IsTrue(left, right EvalResult) (bool, error) {
+	return false, nil
+}
+
+// Type implements the ComparisonOp interface
+func (n *NullSafeEqualOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (n *NullSafeEqualOp) String() string {
+	return "<=>"
+}
+
+// Evaluate implements the ComparisonOp interface
+func (l *LessThanOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	if out, err := l.IsTrue(left, right); err != nil || !out {
+		return resultFalse, err
+	}
+	return resultTrue, nil
+}
+
+// IsTrue implements the ComparisonOp interface
+func (l *LessThanOp) IsTrue(left, right EvalResult) (bool, error) {
+	numeric, err := executeComparison(left, right)
+	if err != nil {
+		return false, err
+	}
+	return numeric < 0, nil
+}
+
+// Type implements the ComparisonOp interface
+func (l *LessThanOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (l *LessThanOp) String() string {
+	return "<"
+}
+
+// Evaluate implements the ComparisonOp interface
+func (l *LessEqualOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	if out, err := l.IsTrue(left, right); err != nil || !out {
+		return resultFalse, err
+	}
+	return resultTrue, nil
+}
+
+// IsTrue implements the ComparisonOp interface
+func (l *LessEqualOp) IsTrue(left, right EvalResult) (bool, error) {
+	numeric, err := executeComparison(left, right)
+	if err != nil {
+		return false, err
+	}
+	return numeric <= 0, nil
+}
+
+// Type implements the ComparisonOp interface
+func (l *LessEqualOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (l *LessEqualOp) String() string {
+	return "<="
+}
+
+// Evaluate implements the ComparisonOp interface
+func (g *GreaterThanOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	if out, err := g.IsTrue(left, right); err != nil || !out {
+		return resultFalse, err
+	}
+	return resultTrue, nil
+}
+
+// IsTrue implements the ComparisonOp interface
+func (g *GreaterThanOp) IsTrue(left, right EvalResult) (bool, error) {
+	numeric, err := executeComparison(left, right)
+	if err != nil {
+		return false, err
+	}
+	return numeric > 0, nil
+}
+
+// Type implements the ComparisonOp interface
+func (g *GreaterThanOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (g *GreaterThanOp) String() string {
+	return ">"
+}
+
+// Evaluate implements the ComparisonOp interface
+func (g *GreaterEqualOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	if out, err := g.IsTrue(left, right); err != nil || !out {
+		return resultFalse, err
+	}
+	return resultTrue, nil
+}
+
+// IsTrue implements the ComparisonOp interface
+func (g *GreaterEqualOp) IsTrue(left, right EvalResult) (bool, error) {
+	numeric, err := executeComparison(left, right)
+	if err != nil {
+		return false, err
+	}
+	return numeric >= 0, nil
+}
+
+// Type implements the ComparisonOp interface
+func (g *GreaterEqualOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (g *GreaterEqualOp) String() string {
+	return ">="
+}
+
+// Evaluate implements the ComparisonOp interface
+func (i *InOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	panic("implement me")
+}
+
+// IsTrue implements the ComparisonOp interface
+func (i *InOp) IsTrue(left, right EvalResult) (bool, error) {
+	return false, nil
+}
+
+// Type implements the ComparisonOp interface
+func (i *InOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (i *InOp) String() string {
+	return "in"
+}
+
+// Evaluate implements the ComparisonOp interface
+func (n *NotInOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	panic("implement me")
+}
+
+// IsTrue implements the ComparisonOp interface
+func (n *NotInOp) IsTrue(left, right EvalResult) (bool, error) {
+	return false, nil
+}
+
+// Type implements the ComparisonOp interface
+func (n *NotInOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (n *NotInOp) String() string {
+	return "not in"
+}
+
+// Evaluate implements the ComparisonOp interface
+func (l *LikeOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	panic("implement me")
+}
+
+// IsTrue implements the ComparisonOp interface
+func (l *LikeOp) IsTrue(left, right EvalResult) (bool, error) {
+	return false, nil
+}
+
+// Type implements the ComparisonOp interface
+func (l *LikeOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (l *LikeOp) String() string {
+	return "like"
+}
+
+// Evaluate implements the ComparisonOp interface
+func (n *NotLikeOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	panic("implement me")
+}
+
+// IsTrue implements the ComparisonOp interface
+func (n *NotLikeOp) IsTrue(left, right EvalResult) (bool, error) {
+	return false, nil
+}
+
+// Type implements the ComparisonOp interface
+func (n *NotLikeOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (n *NotLikeOp) String() string {
+	return "not like"
+}
+
+// Evaluate implements the ComparisonOp interface
+func (r *RegexpOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	panic("implement me")
+}
+
+// IsTrue implements the ComparisonOp interface
+func (r *RegexpOp) IsTrue(left, right EvalResult) (bool, error) {
+	return false, nil
+}
+
+// Type implements the ComparisonOp interface
+func (r *RegexpOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (r *RegexpOp) String() string {
+	return "regexp"
+}
+
+// Evaluate implements the ComparisonOp interface
+func (n *NotRegexpOp) Evaluate(left, right EvalResult) (EvalResult, error) {
+	panic("implement me")
+}
+
+// IsTrue implements the ComparisonOp interface
+func (n *NotRegexpOp) IsTrue(left, right EvalResult) (bool, error) {
+	return false, nil
+}
+
+// Type implements the ComparisonOp interface
+func (n *NotRegexpOp) Type() querypb.Type {
+	return querypb.Type_INT32
+}
+
+// String implements the ComparisonOp interface
+func (n *NotRegexpOp) String() string {
+	return "not regexp"
+}
diff --git a/go/vt/vtgate/evalengine/convert.go b/go/vt/vtgate/evalengine/convert.go
new file mode 100644
index 0000000000..2979a1ac29
--- /dev/null
+++ b/go/vt/vtgate/evalengine/convert.go
@@ -0,0 +1,158 @@
+/*
+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 (
+	"vitess.io/vitess/go/mysql/collations"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/vterrors"
+)
+
+type (
+	ConverterLookup interface {
+		ColumnLookup(col *sqlparser.ColName) (int, error)
+		CollationIDLookup(expr sqlparser.Expr) collations.ID
+	}
+)
+
+var ErrConvertExprNotSupported = "expr cannot be converted, not supported"
+
+// translateComparisonOperator takes in a sqlparser.ComparisonExprOperator and
+// returns the corresponding ComparisonOp
+func translateComparisonOperator(op sqlparser.ComparisonExprOperator) ComparisonOp {
+	switch op {
+	case sqlparser.EqualOp:
+		return &EqualOp{}
+	case sqlparser.LessThanOp:
+		return &LessThanOp{}
+	case sqlparser.GreaterThanOp:
+		return &GreaterThanOp{}
+	case sqlparser.LessEqualOp:
+		return &LessEqualOp{}
+	case sqlparser.GreaterEqualOp:
+		return &GreaterEqualOp{}
+	case sqlparser.NotEqualOp:
+		return &NotEqualOp{}
+	case sqlparser.NullSafeEqualOp:
+		return &NullSafeEqualOp{}
+	case sqlparser.InOp:
+		return &InOp{}
+	case sqlparser.NotInOp:
+		return &NotInOp{}
+	case sqlparser.LikeOp:
+		return &LikeOp{}
+	case sqlparser.NotLikeOp:
+		return &NotLikeOp{}
+	case sqlparser.RegexpOp:
+		return &RegexpOp{}
+	case sqlparser.NotRegexpOp:
+		return &NotRegexpOp{}
+	default:
+		return nil
+	}
+}
+
+func getCollation(expr sqlparser.Expr, lookup ConverterLookup) collations.ID {
+	collation := collations.Unknown
+	if lookup != nil {
+		collation = lookup.CollationIDLookup(expr)
+	}
+	return collation
+}
+
+// Convert converts between AST expressions and executable expressions
+func Convert(e sqlparser.Expr, lookup ConverterLookup) (Expr, error) {
+	switch node := e.(type) {
+	case *sqlparser.ColName:
+		if lookup == nil {
+			return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "%s: cannot lookup column", ErrConvertExprNotSupported)
+		}
+		idx, err := lookup.ColumnLookup(node)
+		if err != nil {
+			return nil, err
+		}
+		collation := getCollation(node, lookup)
+		return NewColumn(idx, collation), nil
+	case *sqlparser.ComparisonExpr:
+		left, err := Convert(node.Left, lookup)
+		if err != nil {
+			return nil, err
+		}
+		right, err := Convert(node.Right, lookup)
+		if err != nil {
+			return nil, err
+		}
+		return &ComparisonExpr{
+			Op:    translateComparisonOperator(node.Operator),
+			Left:  left,
+			Right: right,
+		}, nil
+	case sqlparser.Argument:
+		collation := collations.Unknown
+		if lookup != nil {
+			collation = lookup.CollationIDLookup(e)
+		}
+		return NewBindVar(string(node), collation), nil
+	case *sqlparser.Literal:
+		switch node.Type {
+		case sqlparser.IntVal:
+			return NewLiteralIntFromBytes(node.Bytes())
+		case sqlparser.FloatVal:
+			return NewLiteralFloatFromBytes(node.Bytes())
+		case sqlparser.StrVal:
+			collation := getCollation(e, lookup)
+			return NewLiteralString(node.Bytes(), collation), nil
+		case sqlparser.HexNum:
+			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "%s: hexadecimal value: %s", ErrConvertExprNotSupported, node.Val)
+		}
+	case sqlparser.BoolVal:
+		if node {
+			return NewLiteralIntFromBytes([]byte("1"))
+		}
+		return NewLiteralIntFromBytes([]byte("0"))
+	case *sqlparser.BinaryExpr:
+		var op BinaryOp
+		switch node.Operator {
+		case sqlparser.PlusOp:
+			op = &Addition{}
+		case sqlparser.MinusOp:
+			op = &Subtraction{}
+		case sqlparser.MultOp:
+			op = &Multiplication{}
+		case sqlparser.DivOp:
+			op = &Division{}
+		default:
+			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "%s: %T", ErrConvertExprNotSupported, e)
+		}
+		left, err := Convert(node.Left, lookup)
+		if err != nil {
+			return nil, err
+		}
+		right, err := Convert(node.Right, lookup)
+		if err != nil {
+			return nil, err
+		}
+		return &BinaryExpr{
+			Op:    op,
+			Left:  left,
+			Right: right,
+		}, nil
+
+	}
+	return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "%s: %T", ErrConvertExprNotSupported, e)
+}
diff --git a/go/vt/vtgate/evalengine/evalengine.go b/go/vt/vtgate/evalengine/evalengine.go
index f5e123b5ab..6e51467c95 100644
--- a/go/vt/vtgate/evalengine/evalengine.go
+++ b/go/vt/vtgate/evalengine/evalengine.go
@@ -17,6 +17,10 @@ limitations under the License.
 package evalengine
 
 import (
+	"time"
+
+	"vitess.io/vitess/go/mysql/collations"
+
 	"vitess.io/vitess/go/sqltypes"
 
 	"strconv"
@@ -162,7 +166,11 @@ func newEvalResult(v sqltypes.Value) (EvalResult, error) {
 		if err != nil {
 			return EvalResult{}, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", err)
 		}
-		return EvalResult{fval: fval, typ: sqltypes.Float64}, nil
+		typ := sqltypes.Float64
+		if v.Type() == sqltypes.Decimal {
+			typ = sqltypes.Decimal
+		}
+		return EvalResult{fval: fval, typ: typ}, nil
 	default:
 		return EvalResult{typ: v.Type(), bytes: raw}, nil
 	}
@@ -253,7 +261,8 @@ func hashCode(v EvalResult) int64 {
 }
 
 func compareNumeric(v1, v2 EvalResult) (int, error) {
-	// Equalize the types.
+	// Equalize the types the same way MySQL does
+	// https://dev.mysql.com/doc/refman/8.0/en/type-conversion.html
 	switch v1.typ {
 	case sqltypes.Int64:
 		switch v2.typ {
@@ -262,8 +271,8 @@ func compareNumeric(v1, v2 EvalResult) (int, error) {
 				return -1, nil
 			}
 			v1 = EvalResult{typ: sqltypes.Uint64, uval: uint64(v1.ival)}
-		case sqltypes.Float64:
-			v1 = EvalResult{typ: sqltypes.Float64, fval: float64(v1.ival)}
+		case sqltypes.Float64, sqltypes.Decimal:
+			v1 = EvalResult{typ: v2.typ, fval: float64(v1.ival)}
 		}
 	case sqltypes.Uint64:
 		switch v2.typ {
@@ -272,15 +281,32 @@ func compareNumeric(v1, v2 EvalResult) (int, error) {
 				return 1, nil
 			}
 			v2 = EvalResult{typ: sqltypes.Uint64, uval: uint64(v2.ival)}
-		case sqltypes.Float64:
-			v1 = EvalResult{typ: sqltypes.Float64, fval: float64(v1.uval)}
+		case sqltypes.Float64, sqltypes.Decimal:
+			v1 = EvalResult{typ: v2.typ, fval: float64(v1.uval)}
 		}
 	case sqltypes.Float64:
 		switch v2.typ {
 		case sqltypes.Int64:
 			v2 = EvalResult{typ: sqltypes.Float64, fval: float64(v2.ival)}
 		case sqltypes.Uint64:
+			if v1.fval < 0 {
+				return -1, nil
+			}
 			v2 = EvalResult{typ: sqltypes.Float64, fval: float64(v2.uval)}
+		case sqltypes.Decimal:
+			v2.typ = sqltypes.Float64
+		}
+	case sqltypes.Decimal:
+		switch v2.typ {
+		case sqltypes.Int64:
+			v2 = EvalResult{typ: sqltypes.Decimal, fval: float64(v2.ival)}
+		case sqltypes.Uint64:
+			if v1.fval < 0 {
+				return -1, nil
+			}
+			v2 = EvalResult{typ: sqltypes.Decimal, fval: float64(v2.uval)}
+		case sqltypes.Float64:
+			v1.typ = sqltypes.Float64
 		}
 	}
 
@@ -300,7 +326,7 @@ func compareNumeric(v1, v2 EvalResult) (int, error) {
 		case v1.uval < v2.uval:
 			return -1, nil
 		}
-	case sqltypes.Float64:
+	case sqltypes.Float64, sqltypes.Decimal:
 		switch {
 		case v1.fval == v2.fval:
 			return 0, nil
@@ -312,3 +338,113 @@ func compareNumeric(v1, v2 EvalResult) (int, error) {
 	// v1>v2
 	return 1, nil
 }
+
+func parseDate(expr EvalResult) (t time.Time, err error) {
+	switch expr.typ {
+	case sqltypes.Date:
+		t, err = time.Parse("2006-01-02", string(expr.bytes))
+	case sqltypes.Timestamp, sqltypes.Datetime:
+		t, err = time.Parse("2006-01-02 15:04:05", string(expr.bytes))
+	case sqltypes.Time:
+		t, err = time.Parse("15:04:05", string(expr.bytes))
+		if err == nil {
+			now := time.Now()
+			// setting the date to today's date, because we use AddDate on t
+			// which is "0000-01-01 xx:xx:xx", we do minus one on the month
+			// and day to take into account the 01 in both month and day of t
+			t = t.AddDate(now.Year(), int(now.Month()-1), now.Day()-1)
+		}
+	}
+	return
+}
+
+// matchExprWithAnyDateFormat formats the given expr (usually a string) to a date using the first format
+// that does not return an error.
+func matchExprWithAnyDateFormat(expr EvalResult) (t time.Time, err error) {
+	layouts := []string{"2006-01-02", "2006-01-02 15:04:05", "15:04:05"}
+	for _, layout := range layouts {
+		t, err = time.Parse(layout, string(expr.bytes))
+		if err == nil {
+			if layout == "15:04:05" {
+				now := time.Now()
+				// setting the date to today's date, because we use AddDate on t
+				// which is "0000-01-01 xx:xx:xx", we do minus one on the month
+				// and day to take into account the 01 in both month and day of t
+				t = t.AddDate(now.Year(), int(now.Month()-1), now.Day()-1)
+			}
+			return
+		}
+	}
+	return
+}
+
+// Date comparison based on:
+// 		- https://dev.mysql.com/doc/refman/8.0/en/type-conversion.html
+// 		- https://dev.mysql.com/doc/refman/8.0/en/date-and-time-type-conversion.html
+func compareDates(l, r EvalResult) (int, error) {
+	lTime, err := parseDate(l)
+	if err != nil {
+		return 0, err
+	}
+	rTime, err := parseDate(r)
+	if err != nil {
+		return 0, err
+	}
+
+	return compareGoTimes(lTime, rTime)
+}
+
+func compareDateAndString(l, r EvalResult) (int, error) {
+	var lTime, rTime time.Time
+	var err error
+	switch {
+	case sqltypes.IsDate(l.typ):
+		lTime, err = parseDate(l)
+		if err != nil {
+			return 0, err
+		}
+		rTime, err = matchExprWithAnyDateFormat(r)
+		if err != nil {
+			return 0, err
+		}
+	case sqltypes.IsText(l.typ) || sqltypes.IsBinary(l.typ):
+		rTime, err = parseDate(r)
+		if err != nil {
+			return 0, err
+		}
+		lTime, err = matchExprWithAnyDateFormat(l)
+		if err != nil {
+			return 0, err
+		}
+	}
+	return compareGoTimes(lTime, rTime)
+}
+
+func compareGoTimes(lTime, rTime time.Time) (int, error) {
+	if lTime.Before(rTime) {
+		return -1, nil
+	}
+	if lTime.After(rTime) {
+		return 1, nil
+	}
+	return 0, nil
+}
+
+// More on string collations coercibility on MySQL documentation:
+// 		- https://dev.mysql.com/doc/refman/8.0/en/charset-collation-coercibility.html
+func compareStrings(l, r EvalResult) (int, error) {
+	// If one of the strings has an unknown collation we fail, though such error should
+	// already be handled before the execution by the planner.
+	if l.collation == collations.Unknown || r.collation == collations.Unknown {
+		return 0, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "cannot compare strings with an unknown collation")
+	}
+
+	// We cannot compare different collations for now, so we fail
+	// TODO: support multiple collations comparison
+	if r.collation != l.collation {
+		return 0, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "cannot compare strings with different collations")
+	}
+
+	collation := collations.Default().LookupByID(l.collation)
+	return collation.Collate(l.bytes, r.bytes, false), nil
+}
diff --git a/go/vt/vtgate/evalengine/expressions.go b/go/vt/vtgate/evalengine/expressions.go
index a32badee48..6f9e7503cf 100644
--- a/go/vt/vtgate/evalengine/expressions.go
+++ b/go/vt/vtgate/evalengine/expressions.go
@@ -20,6 +20,8 @@ import (
 	"fmt"
 	"strconv"
 
+	"vitess.io/vitess/go/mysql/collations"
+
 	"vitess.io/vitess/go/sqltypes"
 
 	querypb "vitess.io/vitess/go/vt/proto/query"
@@ -29,14 +31,16 @@ import (
 
 type (
 	EvalResult struct {
-		typ   querypb.Type
-		ival  int64
-		uval  uint64
-		fval  float64
-		bytes []byte
+		typ       querypb.Type
+		ival      int64
+		uval      uint64
+		fval      float64
+		bytes     []byte
+		collation collations.ID
 	}
-	//ExpressionEnv contains the environment that the expression
-	//evaluates in, such as the current row and bindvars
+
+	// ExpressionEnv contains the environment that the expression
+	// evaluates in, such as the current row and bindvars
 	ExpressionEnv struct {
 		BindVars map[string]*querypb.BindVariable
 		Row      []sqltypes.Value
@@ -49,35 +53,35 @@ type (
 		String() string
 	}
 
-	//BinaryExpr allows binary expressions to not have to evaluate child expressions - this is done by the BinaryOp
-	BinaryExpr interface {
-		Evaluate(left, right EvalResult) (EvalResult, error)
-		Type(left querypb.Type) querypb.Type
-		String() string
-	}
-
 	// Expressions
-	Literal      struct{ Val EvalResult }
-	BindVariable struct{ Key string }
-	Column       struct{ Offset int }
-	BinaryOp     struct {
-		Expr        BinaryExpr
-		Left, Right Expr
+	Null    struct{}
+	Literal struct {
+		Val       EvalResult
+		Collation collations.ID
+	}
+	BindVariable struct {
+		Key       string
+		Collation collations.ID
+	}
+	Column struct {
+		Offset    int
+		Collation collations.ID
 	}
-
-	// Binary ops
-	Addition       struct{}
-	Subtraction    struct{}
-	Multiplication struct{}
-	Division       struct{}
 )
 
-//Value allows for retrieval of the value we expose for public consumption
+var _ Expr = (*Null)(nil)
+var _ Expr = (*Literal)(nil)
+var _ Expr = (*BindVariable)(nil)
+var _ Expr = (*Column)(nil)
+var _ Expr = (*BinaryExpr)(nil)
+var _ Expr = (*ComparisonExpr)(nil)
+
+// Value allows for retrieval of the value we expose for public consumption
 func (e EvalResult) Value() sqltypes.Value {
 	return e.toSQLValue(e.typ)
 }
 
-//NewLiteralIntFromBytes returns a literal expression
+// NewLiteralIntFromBytes returns a literal expression
 func NewLiteralIntFromBytes(val []byte) (Expr, error) {
 	ival, err := strconv.ParseInt(string(val), 10, 64)
 	if err != nil {
@@ -86,136 +90,91 @@ func NewLiteralIntFromBytes(val []byte) (Expr, error) {
 	return NewLiteralInt(ival), nil
 }
 
-//NewLiteralInt returns a literal expression
+// NewLiteralInt returns a literal expression
 func NewLiteralInt(i int64) Expr {
-	return &Literal{EvalResult{typ: sqltypes.Int64, ival: i}}
+	return &Literal{Val: EvalResult{typ: sqltypes.Int64, ival: i}}
 }
 
-//NewLiteralFloat returns a literal expression
-func NewLiteralFloat(val []byte) (Expr, error) {
+// NewLiteralFloat returns a literal expression
+func NewLiteralFloat(val float64) Expr {
+	return &Literal{Val: EvalResult{typ: sqltypes.Float64, fval: val}}
+}
+
+// NewLiteralFloatFromBytes returns a float literal expression from a slice of bytes
+func NewLiteralFloatFromBytes(val []byte) (Expr, error) {
 	fval, err := strconv.ParseFloat(string(val), 64)
 	if err != nil {
 		return nil, err
 	}
-	return &Literal{EvalResult{typ: sqltypes.Float64, fval: fval}}, nil
+	return &Literal{Val: EvalResult{typ: sqltypes.Float64, fval: fval}}, nil
 }
 
-//NewLiteralFloat returns a literal expression
-func NewLiteralString(val []byte) Expr {
-	return &Literal{EvalResult{typ: sqltypes.VarBinary, bytes: val}}
+// NewLiteralString returns a literal expression
+func NewLiteralString(val []byte, collation collations.ID) Expr {
+	return &Literal{Val: EvalResult{typ: sqltypes.VarBinary, bytes: val}, Collation: collation}
 }
 
-//NewBindVar returns a bind variable
-func NewBindVar(key string) Expr {
-	return &BindVariable{Key: key}
+// NewBindVar returns a bind variable
+func NewBindVar(key string, collation collations.ID) Expr {
+	return &BindVariable{
+		Key:       key,
+		Collation: collation,
+	}
 }
 
-//NewColumn returns a bind variable
-func NewColumn(offset int) Expr {
+// NewColumn returns a bind variable
+func NewColumn(offset int, collation collations.ID) Expr {
 	return &Column{
-		Offset: offset,
+		Offset:    offset,
+		Collation: collation,
 	}
 }
 
-var _ Expr = (*Literal)(nil)
-var _ Expr = (*BindVariable)(nil)
-var _ Expr = (*BinaryOp)(nil)
-var _ Expr = (*Column)(nil)
+// Evaluate implements the Expr interface
+func (n Null) Evaluate(ExpressionEnv) (EvalResult, error) {
+	return EvalResult{}, nil
+}
 
-var _ BinaryExpr = (*Addition)(nil)
-var _ BinaryExpr = (*Subtraction)(nil)
-var _ BinaryExpr = (*Multiplication)(nil)
-var _ BinaryExpr = (*Division)(nil)
+// Type implements the Expr interface
+func (n Null) Type(ExpressionEnv) (querypb.Type, error) {
+	return querypb.Type_NULL_TYPE, nil
+}
 
-//Evaluate implements the Expr interface
-func (b *BinaryOp) Evaluate(env ExpressionEnv) (EvalResult, error) {
-	lVal, err := b.Left.Evaluate(env)
-	if err != nil {
-		return EvalResult{}, err
-	}
-	rVal, err := b.Right.Evaluate(env)
-	if err != nil {
-		return EvalResult{}, err
-	}
-	return b.Expr.Evaluate(lVal, rVal)
+// String implements the Expr interface
+func (n Null) String() string {
+	return "null"
 }
 
-//Evaluate implements the Expr interface
+// Evaluate implements the Expr interface
 func (l *Literal) Evaluate(ExpressionEnv) (EvalResult, error) {
-	return l.Val, nil
+	eval := l.Val
+	eval.collation = l.Collation
+	return eval, nil
 }
 
-//Evaluate implements the Expr interface
+// Evaluate implements the Expr interface
 func (b *BindVariable) Evaluate(env ExpressionEnv) (EvalResult, error) {
 	val, ok := env.BindVars[b.Key]
 	if !ok {
 		return EvalResult{}, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "Bind variable not found")
 	}
-	return evaluateByType(val)
+	eval, err := evaluateByType(val)
+	if err != nil {
+		return EvalResult{}, err
+	}
+	eval.collation = b.Collation
+	return eval, nil
 }
 
-//Evaluate implements the Expr interface
+// Evaluate implements the Expr interface
 func (c *Column) Evaluate(env ExpressionEnv) (EvalResult, error) {
 	value := env.Row[c.Offset]
 	numeric, err := newEvalResult(value)
+	numeric.collation = c.Collation
 	return numeric, err
 }
 
-//Evaluate implements the BinaryOp interface
-func (a *Addition) Evaluate(left, right EvalResult) (EvalResult, error) {
-	return addNumericWithError(left, right)
-}
-
-//Evaluate implements the BinaryOp interface
-func (s *Subtraction) Evaluate(left, right EvalResult) (EvalResult, error) {
-	return subtractNumericWithError(left, right)
-}
-
-//Evaluate implements the BinaryOp interface
-func (m *Multiplication) Evaluate(left, right EvalResult) (EvalResult, error) {
-	return multiplyNumericWithError(left, right)
-}
-
-//Evaluate implements the BinaryOp interface
-func (d *Division) Evaluate(left, right EvalResult) (EvalResult, error) {
-	return divideNumericWithError(left, right)
-}
-
-//Type implements the BinaryExpr interface
-func (a *Addition) Type(left querypb.Type) querypb.Type {
-	return left
-}
-
-//Type implements the BinaryExpr interface
-func (m *Multiplication) Type(left querypb.Type) querypb.Type {
-	return left
-}
-
-//Type implements the BinaryExpr interface
-func (d *Division) Type(querypb.Type) querypb.Type {
-	return sqltypes.Float64
-}
-
-//Type implements the BinaryExpr interface
-func (s *Subtraction) Type(left querypb.Type) querypb.Type {
-	return left
-}
-
-//Type implements the Expr interface
-func (b *BinaryOp) Type(env ExpressionEnv) (querypb.Type, error) {
-	ltype, err := b.Left.Type(env)
-	if err != nil {
-		return 0, err
-	}
-	rtype, err := b.Right.Type(env)
-	if err != nil {
-		return 0, err
-	}
-	typ := mergeNumericalTypes(ltype, rtype)
-	return b.Expr.Type(typ), nil
-}
-
-//Type implements the Expr interface
+// Type implements the Expr interface
 func (b *BindVariable) Type(env ExpressionEnv) (querypb.Type, error) {
 	e := env.BindVars
 	v, found := e[b.Key]
@@ -225,52 +184,27 @@ func (b *BindVariable) Type(env ExpressionEnv) (querypb.Type, error) {
 	return v.Type, nil
 }
 
-//Type implements the Expr interface
+// Type implements the Expr interface
 func (l *Literal) Type(ExpressionEnv) (querypb.Type, error) {
 	return l.Val.typ, nil
 }
 
-//Type implements the Expr interface
+// Type implements the Expr interface
 func (c *Column) Type(ExpressionEnv) (querypb.Type, error) {
 	return sqltypes.Float64, nil
 }
 
-//String implements the BinaryExpr interface
-func (d *Division) String() string {
-	return "/"
-}
-
-//String implements the BinaryExpr interface
-func (m *Multiplication) String() string {
-	return "*"
-}
-
-//String implements the BinaryExpr interface
-func (s *Subtraction) String() string {
-	return "-"
-}
-
-//String implements the BinaryExpr interface
-func (a *Addition) String() string {
-	return "+"
-}
-
-//String implements the Expr interface
-func (b *BinaryOp) String() string {
-	return b.Left.String() + " " + b.Expr.String() + " " + b.Right.String()
-}
-
-//String implements the Expr interface
+// String implements the Expr interface
 func (b *BindVariable) String() string {
 	return ":" + b.Key
 }
 
-//String implements the Expr interface
+// String implements the Expr interface
 func (l *Literal) String() string {
 	return l.Val.Value().String()
 }
 
-//String implements the Expr interface
+// String implements the Expr interface
 func (c *Column) String() string {
 	return fmt.Sprintf("column %d from the input", c.Offset)
 }
@@ -317,13 +251,15 @@ func evaluateByType(val *querypb.BindVariable) (EvalResult, error) {
 		return EvalResult{typ: sqltypes.Float64, fval: fval}, nil
 	case sqltypes.VarChar, sqltypes.Text, sqltypes.VarBinary:
 		return EvalResult{typ: sqltypes.VarBinary, bytes: val.Value}, nil
+	case sqltypes.Time, sqltypes.Datetime, sqltypes.Timestamp, sqltypes.Date:
+		return EvalResult{typ: val.Type, bytes: val.Value}, nil
 	case sqltypes.Null:
 		return EvalResult{typ: sqltypes.Null}, nil
 	}
 	return EvalResult{}, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "Type is not supported: %s", val.Type.String())
 }
 
-// debugString is
+// debugString prints the entire EvalResult in a debug format
 func (e *EvalResult) debugString() string {
 	return fmt.Sprintf("(%s) %d %d %f %s", querypb.Type_name[int32(e.typ)], e.ival, e.uval, e.fval, string(e.bytes))
 }
diff --git a/go/vt/vtgate/planbuilder/expression_converter.go b/go/vt/vtgate/planbuilder/expression_converter.go
index 769c4b6f52..1bcde04bbc 100644
--- a/go/vt/vtgate/planbuilder/expression_converter.go
+++ b/go/vt/vtgate/planbuilder/expression_converter.go
@@ -62,7 +62,7 @@ func booleanValues(astExpr sqlparser.Expr) evalengine.Expr {
 func identifierAsStringValue(astExpr sqlparser.Expr) evalengine.Expr {
 	colName, isColName := astExpr.(*sqlparser.ColName)
 	if isColName {
-		return evalengine.NewLiteralString([]byte(colName.Name.Lowered()))
+		return evalengine.NewLiteralString([]byte(colName.Name.Lowered()), 0)
 	}
 	return nil
 }
@@ -80,9 +80,9 @@ func (ec *expressionConverter) convert(astExpr sqlparser.Expr, boolean, identifi
 			return evalExpr, nil
 		}
 	}
-	evalExpr, err := sqlparser.Convert(astExpr)
+	evalExpr, err := evalengine.Convert(astExpr, nil)
 	if err != nil {
-		if err != sqlparser.ErrExprNotSupported {
+		if !strings.Contains(err.Error(), evalengine.ErrConvertExprNotSupported) {
 			return nil, err
 		}
 		evalExpr = &evalengine.Column{Offset: len(ec.tabletExpressions)}
diff --git a/go/vt/vtgate/planbuilder/filter.go b/go/vt/vtgate/planbuilder/filter.go
new file mode 100644
index 0000000000..1ab52a075d
--- /dev/null
+++ b/go/vt/vtgate/planbuilder/filter.go
@@ -0,0 +1,83 @@
+/*
+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 planbuilder
+
+import (
+	"vitess.io/vitess/go/mysql/collations"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/vterrors"
+	"vitess.io/vitess/go/vt/vtgate/engine"
+	"vitess.io/vitess/go/vt/vtgate/evalengine"
+	"vitess.io/vitess/go/vt/vtgate/semantics"
+)
+
+type (
+	// filter is the logicalPlan for engine.Filter.
+	filter struct {
+		logicalPlanCommon
+		efilter *engine.Filter
+	}
+
+	simpleConverterLookup struct {
+		semTable *semantics.SemTable
+		plan     logicalPlan
+	}
+)
+
+var _ logicalPlan = (*filter)(nil)
+var _ evalengine.ConverterLookup = (*simpleConverterLookup)(nil)
+
+func (s *simpleConverterLookup) ColumnLookup(col *sqlparser.ColName) (int, error) {
+	offset, added, err := pushProjection(&sqlparser.AliasedExpr{Expr: col}, s.plan, s.semTable, true, true, false)
+	if err != nil {
+		return 0, err
+	}
+	if added {
+		return 0, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "column should not be pushed to projection while doing a column lookup")
+	}
+	return offset, nil
+}
+
+func (s *simpleConverterLookup) CollationIDLookup(expr sqlparser.Expr) collations.ID {
+	return s.semTable.CollationFor(expr)
+}
+
+// newFilter builds a new filter.
+func newFilter(semTable *semantics.SemTable, plan logicalPlan, expr sqlparser.Expr) (*filter, error) {
+	scl := &simpleConverterLookup{
+		semTable: semTable,
+		plan:     plan,
+	}
+	predicate, err := evalengine.Convert(expr, scl)
+	if err != nil {
+		return nil, err
+	}
+	return &filter{
+		logicalPlanCommon: newBuilderCommon(plan),
+		efilter: &engine.Filter{
+			Predicate:    predicate,
+			ASTPredicate: expr,
+		},
+	}, nil
+}
+
+// Primitive implements the logicalPlan interface
+func (l *filter) Primitive() engine.Primitive {
+	l.efilter.Input = l.input.Primitive()
+	return l.efilter
+}
diff --git a/go/vt/vtgate/planbuilder/horizon_planning.go b/go/vt/vtgate/planbuilder/horizon_planning.go
index 4e748409ec..368a241dd3 100644
--- a/go/vt/vtgate/planbuilder/horizon_planning.go
+++ b/go/vt/vtgate/planbuilder/horizon_planning.go
@@ -147,6 +147,8 @@ func (hp *horizonPlanning) truncateColumnsIfNeeded(plan logicalPlan) error {
 		p.truncater.SetTruncateColumnCount(hp.sel.GetColumnCount())
 	case *pulloutSubquery:
 		return hp.truncateColumnsIfNeeded(p.underlying)
+	case *filter:
+		return hp.truncateColumnsIfNeeded(p.input)
 	default:
 		return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "plan type not known for column truncation: %T", plan)
 	}
@@ -458,7 +460,7 @@ func (hp *horizonPlanning) planAggregations(ctx *planningContext, plan logicalPl
 		hp.haveToTruncate(added)
 	}
 
-	err := hp.planHaving(ctx, newPlan)
+	newPlan, err := hp.planHaving(ctx, newPlan)
 	if err != nil {
 		return nil, err
 	}
@@ -660,35 +662,28 @@ func (hp *horizonPlanning) planOrderBy(ctx *planningContext, orderExprs []abstra
 		return plan, nil
 	case *memorySort:
 		return plan, nil
-	case *pulloutSubquery:
-		newUnderlyingPlan, err := hp.planOrderBy(ctx, orderExprs, plan.underlying)
-		if err != nil {
-			return nil, err
+	case *simpleProjection:
+		return hp.createMemorySortPlan(ctx, plan, orderExprs, true)
+	case *vindexFunc:
+		// This is evaluated at VTGate only, so weight_string function cannot be used.
+		return hp.createMemorySortPlan(ctx, plan, orderExprs /* useWeightStr */, false)
+	case *limit, *semiJoin, *filter, *pulloutSubquery:
+		inputs := plan.Inputs()
+		if len(inputs) == 0 {
+			break
 		}
-		plan.underlying = newUnderlyingPlan
-		return plan, nil
-	case *semiJoin:
-		newUnderlyingPlan, err := hp.planOrderBy(ctx, orderExprs, plan.lhs)
+		newFirstInput, err := hp.planOrderBy(ctx, orderExprs, inputs[0])
 		if err != nil {
 			return nil, err
 		}
-		plan.lhs = newUnderlyingPlan
-		return plan, nil
-	case *limit:
-		newUnderlyingPlan, err := hp.planOrderBy(ctx, orderExprs, plan.input)
+		inputs[0] = newFirstInput
+		err = plan.Rewrite(inputs...)
 		if err != nil {
 			return nil, err
 		}
-		plan.input = newUnderlyingPlan
 		return plan, nil
-	case *simpleProjection:
-		return hp.createMemorySortPlan(ctx, plan, orderExprs, true)
-	case *vindexFunc:
-		// This is evaluated at VTGate only, so weight_string function cannot be used.
-		return hp.createMemorySortPlan(ctx, plan, orderExprs /* useWeightStr */, false)
-	default:
-		return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "ordering on complex query %T", plan)
 	}
+	return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "ordering on complex query %T", plan)
 }
 
 func isSpecialOrderBy(o abstract.OrderBy) bool {
@@ -1082,31 +1077,25 @@ func (hp *horizonPlanning) needDistinctHandling(ctx *planningContext, funcExpr *
 	return true, innerAliased, nil
 }
 
-func (hp *horizonPlanning) planHaving(ctx *planningContext, plan logicalPlan) error {
+func (hp *horizonPlanning) planHaving(ctx *planningContext, plan logicalPlan) (logicalPlan, error) {
 	if hp.sel.Having == nil {
-		return nil
-	}
-	for _, expr := range sqlparser.SplitAndExpression(nil, hp.sel.Having.Expr) {
-		err := pushHaving(expr, plan, ctx.semTable)
-		if err != nil {
-			return err
-		}
+		return plan, nil
 	}
-	return nil
+	return pushHaving(hp.sel.Having.Expr, plan, ctx.semTable)
 }
 
-func pushHaving(expr sqlparser.Expr, plan logicalPlan, semTable *semantics.SemTable) error {
+func pushHaving(expr sqlparser.Expr, plan logicalPlan, semTable *semantics.SemTable) (logicalPlan, error) {
 	switch node := plan.(type) {
 	case *route:
 		sel := sqlparser.GetFirstSelect(node.Select)
 		sel.AddHaving(expr)
-		return nil
+		return plan, nil
 	case *pulloutSubquery:
 		return pushHaving(expr, node.underlying, semTable)
 	case *simpleProjection:
-		return vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: filtering on results of cross-shard derived table")
+		return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: filtering on results of cross-shard derived table")
 	case *orderedAggregate:
-		return vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: filtering on results of aggregates")
+		return newFilter(semTable, plan, expr)
 	}
-	return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "[BUG] unreachable %T.filtering", plan)
+	return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "[BUG] unreachable %T.filtering", plan)
 }
diff --git a/go/vt/vtgate/planbuilder/rewrite.go b/go/vt/vtgate/planbuilder/rewrite.go
index 99d725f5e9..143fcdbc64 100644
--- a/go/vt/vtgate/planbuilder/rewrite.go
+++ b/go/vt/vtgate/planbuilder/rewrite.go
@@ -167,26 +167,36 @@ func rewriteHavingClause(node *sqlparser.Select) {
 		selectExprMap[aliasedExpr.As.Lowered()] = aliasedExpr.Expr
 	}
 
-	sqlparser.Rewrite(node.Having.Expr, func(cursor *sqlparser.Cursor) bool {
-		switch x := cursor.Node().(type) {
-		case *sqlparser.ColName:
-			if !x.Qualifier.IsEmpty() {
-				return false
-			}
-			originalExpr, isInMap := selectExprMap[x.Name.Lowered()]
-			if isInMap {
-				cursor.Replace(originalExpr)
-				return false
-			}
-			return false
-		}
-		return true
-	}, nil)
-
+	// for each expression in the having clause, we check if it contains aggregation.
+	// if it does, we keep the expression in the having clause ; and if it does not
+	// and the expression is in the select list, we replace the expression by the one
+	// used in the select list and add it to the where clause instead of the having clause.
 	exprs := sqlparser.SplitAndExpression(nil, node.Having.Expr)
 	node.Having = nil
 	for _, expr := range exprs {
-		if sqlparser.ContainsAggregation(expr) {
+		var hasAggr bool
+		sqlparser.Rewrite(expr, func(cursor *sqlparser.Cursor) bool {
+			switch x := cursor.Node().(type) {
+			case *sqlparser.ColName:
+				if !x.Qualifier.IsEmpty() {
+					return false
+				}
+				originalExpr, isInMap := selectExprMap[x.Name.Lowered()]
+				if isInMap {
+					if sqlparser.ContainsAggregation(originalExpr) {
+						hasAggr = true
+					} else {
+						cursor.Replace(originalExpr)
+					}
+				}
+				return false
+			default:
+				hasAggr = hasAggr || sqlparser.IsAggregation(x)
+			}
+			return true
+		}, nil)
+
+		if hasAggr {
 			node.AddHaving(expr)
 		} else {
 			node.AddWhere(expr)
diff --git a/go/vt/vtgate/planbuilder/select.go b/go/vt/vtgate/planbuilder/select.go
index 07a9e027f6..9d5cecb11c 100644
--- a/go/vt/vtgate/planbuilder/select.go
+++ b/go/vt/vtgate/planbuilder/select.go
@@ -320,9 +320,8 @@ func handleDualSelects(sel *sqlparser.Select, vschema ContextVSchema) (engine.Pr
 		if sqlparser.IsLockingFunc(expr.Expr) {
 			// if we are using any locking functions, we bail out here and send the whole query to a single destination
 			return buildLockingPrimitive(sel, vschema)
-
 		}
-		exprs[i], err = sqlparser.Convert(expr.Expr)
+		exprs[i], err = evalengine.Convert(expr.Expr, nil)
 		if err != nil {
 			return nil, nil
 		}
diff --git a/go/vt/vtgate/planbuilder/system_tables.go b/go/vt/vtgate/planbuilder/system_tables.go
index a6cc18d01c..3859d73c59 100644
--- a/go/vt/vtgate/planbuilder/system_tables.go
+++ b/go/vt/vtgate/planbuilder/system_tables.go
@@ -17,11 +17,28 @@ limitations under the License.
 package planbuilder
 
 import (
+	"strings"
+
+	"vitess.io/vitess/go/mysql/collations"
+
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/vterrors"
+
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
 )
 
+type notImplementedSchemaInfoConverter struct{}
+
+func (f *notImplementedSchemaInfoConverter) ColumnLookup(*sqlparser.ColName) (int, error) {
+	return 0, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "Comparing table schema name with a column name not yet supported")
+}
+
+func (f *notImplementedSchemaInfoConverter) CollationIDLookup(sqlparser.Expr) collations.ID {
+	return 0
+}
+
 func (pb *primitiveBuilder) findSysInfoRoutingPredicates(expr sqlparser.Expr, rut *route, reservedVars *sqlparser.ReservedVars) error {
 	isTableSchema, bvName, out, err := extractInfoSchemaRoutingPredicate(expr, reservedVars)
 	if err != nil {
@@ -104,9 +121,9 @@ func extractInfoSchemaRoutingPredicate(in sqlparser.Expr, reservedVars *sqlparse
 		if cmp.Operator == sqlparser.EqualOp {
 			isSchemaName, col, other, replaceOther := findOtherComparator(cmp)
 			if col != nil && shouldRewrite(other) {
-				evalExpr, err := sqlparser.Convert(other)
+				evalExpr, err := evalengine.Convert(other, &notImplementedSchemaInfoConverter{})
 				if err != nil {
-					if err == sqlparser.ErrExprNotSupported {
+					if strings.Contains(err.Error(), evalengine.ErrConvertExprNotSupported) {
 						// This just means we can't rewrite this particular expression,
 						// not that we have to exit altogether
 						return false, "", nil, nil
diff --git a/go/vt/vtgate/planbuilder/system_variables.go b/go/vt/vtgate/planbuilder/system_variables.go
index d6e9cbeaa1..86f758d765 100644
--- a/go/vt/vtgate/planbuilder/system_variables.go
+++ b/go/vt/vtgate/planbuilder/system_variables.go
@@ -61,7 +61,7 @@ func parseAndBuildDefaultValue(sysvar sysvars.SystemVariable) evalengine.Expr {
 	}
 	sel := stmt.(*sqlparser.Select)
 	aliasedExpr := sel.SelectExprs[0].(*sqlparser.AliasedExpr)
-	def, err := sqlparser.Convert(aliasedExpr.Expr)
+	def, err := evalengine.Convert(aliasedExpr.Expr, nil)
 	if err != nil {
 		panic(fmt.Sprintf("bug in set plan init - default value for %s not able to convert to evalengine.Expr: %s", sysvar.Name, sysvar.Default))
 	}
diff --git a/go/vt/vtgate/vindexes/cached_size.go b/go/vt/vtgate/vindexes/cached_size.go
index 10af44ae6b..931cf2be12 100644
--- a/go/vt/vtgate/vindexes/cached_size.go
+++ b/go/vt/vtgate/vindexes/cached_size.go
@@ -93,10 +93,12 @@ func (cached *Column) CachedSize(alloc bool) int64 {
 	}
 	size := int64(0)
 	if alloc {
-		size += int64(48)
+		size += int64(64)
 	}
 	// field Name vitess.io/vitess/go/vt/sqlparser.ColIdent
 	size += cached.Name.CachedSize(false)
+	// field CollationName string
+	size += hack.RuntimeAllocSize(int64(len(cached.CollationName)))
 	return size
 }
 func (cached *ColumnVindex) CachedSize(alloc bool) int64 {
@@ -396,7 +398,7 @@ func (cached *Table) CachedSize(alloc bool) int64 {
 	size += cached.AutoIncrement.CachedSize(true)
 	// field Columns []vitess.io/vitess/go/vt/vtgate/vindexes.Column
 	{
-		size += hack.RuntimeAllocSize(int64(cap(cached.Columns)) * int64(44))
+		size += hack.RuntimeAllocSize(int64(cap(cached.Columns)) * int64(64))
 		for _, elem := range cached.Columns {
 			size += elem.CachedSize(false)
 		}


Test patch:
diff --git a/go/test/endtoend/vtgate/queries/aggregation/aggregation_test.go b/go/test/endtoend/vtgate/queries/aggregation/aggregation_test.go
index 3183b5f97a..197892f9ef 100644
--- a/go/test/endtoend/vtgate/queries/aggregation/aggregation_test.go
+++ b/go/test/endtoend/vtgate/queries/aggregation/aggregation_test.go
@@ -18,6 +18,7 @@ package aggregation
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"vitess.io/vitess/go/test/endtoend/vtgate/utils"
@@ -85,3 +86,203 @@ func TestDistinct(t *testing.T) {
 	utils.Exec(t, conn, "delete from t7_xxhash")
 	utils.Exec(t, conn, "delete from aggr_test")
 }
+
+func TestEqualFilterOnScatter(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.Nil(t, err)
+	defer conn.Close()
+
+	utils.Exec(t, conn, "insert into aggr_test(id, val1, val2) values(1,'a',1), (2,'b',2), (3,'c',3), (4,'d',4), (5,'e',5)")
+
+	defer func() {
+		utils.Exec(t, conn, "set workload = 'oltp'")
+		utils.Exec(t, conn, "delete from aggr_test")
+	}()
+
+	workloads := []string{"oltp", "olap"}
+	for _, workload := range workloads {
+		t.Run(workload, func(t *testing.T) {
+			utils.Exec(t, conn, fmt.Sprintf("set workload = '%s'", workload))
+
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having 1 = 1", `[[INT64(5)]]`)
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a = 5", `[[INT64(5)]]`)
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having 5 = a", `[[INT64(5)]]`)
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a = a", `[[INT64(5)]]`)
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a = 3+2", `[[INT64(5)]]`)
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having 1+4 = 3+2", `[[INT64(5)]]`)
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a = 1", `[]`)
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a = \"1\"", `[]`)
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a = \"5\"", `[[INT64(5)]]`)
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a = 5.00", `[[INT64(5)]]`)
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a, val1 from aggr_test group by val1 having a = 1.00", `[[INT64(1) VARCHAR("a")] [INT64(1) VARCHAR("b")] [INT64(1) VARCHAR("c")] [INT64(1) VARCHAR("d")] [INT64(1) VARCHAR("e")]]`)
+
+			utils.AssertContainsError(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ 1 from aggr_test having count(*) = 5", `expr cannot be converted, not supported: *sqlparser.FuncExpr`) // will fail since `count(*)` is a FuncExpr
+		})
+	}
+}
+
+func TestNotEqualFilterOnScatter(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.Nil(t, err)
+	defer conn.Close()
+
+	utils.Exec(t, conn, "insert into aggr_test(id, val1, val2) values(1,'a',1), (2,'b',2), (3,'c',3), (4,'d',4), (5,'e',5)")
+
+	defer func() {
+		utils.Exec(t, conn, "set workload = oltp")
+		utils.Exec(t, conn, "delete from aggr_test")
+	}()
+
+	workloads := []string{"oltp", "olap"}
+	for _, workload := range workloads {
+		t.Run(workload, func(t *testing.T) {
+			utils.Exec(t, conn, fmt.Sprintf("set workload = '%s'", workload))
+
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a != 5", `[]`)               // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having 5 != a", `[]`)               // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a != a", `[]`)               // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a != 3+2", `[]`)             // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a != 1", `[[INT64(5)]]`)     // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a != \"1\"", `[[INT64(5)]]`) // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a != \"5\"", `[]`)           // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a != 5.00", `[]`)            // having clause
+
+			utils.AssertContainsError(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ 1 from aggr_test having count(*) != 5", `expr cannot be converted, not supported: *sqlparser.FuncExpr`) // will fail since `count(*)` is a FuncExpr
+		})
+	}
+}
+
+func TestLessFilterOnScatter(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.Nil(t, err)
+	defer conn.Close()
+
+	utils.Exec(t, conn, "insert into aggr_test(id, val1, val2) values(1,'a',1), (2,'b',2), (3,'c',3), (4,'d',4), (5,'e',5)")
+
+	defer func() {
+		utils.Exec(t, conn, "set workload = oltp")
+		utils.Exec(t, conn, "delete from aggr_test")
+	}()
+
+	workloads := []string{"oltp", "olap"}
+	for _, workload := range workloads {
+		t.Run(workload, func(t *testing.T) {
+			utils.Exec(t, conn, fmt.Sprintf("set workload = '%s'", workload))
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a < 10", `[[INT64(5)]]`)     // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having 1 < a", `[[INT64(5)]]`)      // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a < a", `[]`)                // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a < 3+2", `[]`)              // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a < 1", `[]`)                // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a < \"10\"", `[[INT64(5)]]`) // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a < \"5\"", `[]`)            // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a < 6.00", `[[INT64(5)]]`)   // having clause
+
+			utils.AssertContainsError(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ 1 from aggr_test having count(*) < 5", `expr cannot be converted, not supported: *sqlparser.FuncExpr`) // will fail since `count(*)` is a FuncExpr
+		})
+	}
+}
+
+func TestLessEqualFilterOnScatter(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.Nil(t, err)
+	defer conn.Close()
+
+	utils.Exec(t, conn, "insert into aggr_test(id, val1, val2) values(1,'a',1), (2,'b',2), (3,'c',3), (4,'d',4), (5,'e',5)")
+
+	defer func() {
+		utils.Exec(t, conn, "set workload = oltp")
+		utils.Exec(t, conn, "delete from aggr_test")
+	}()
+
+	workloads := []string{"oltp", "olap"}
+	for _, workload := range workloads {
+		t.Run(workload, func(t *testing.T) {
+			utils.Exec(t, conn, fmt.Sprintf("set workload = '%s'", workload))
+
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a <= 10", `[[INT64(5)]]`)     // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having 1 <= a", `[[INT64(5)]]`)      // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a <= a", `[[INT64(5)]]`)      // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a <= 3+2", `[[INT64(5)]]`)    // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a <= 1", `[]`)                // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a <= \"10\"", `[[INT64(5)]]`) // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a <= \"5\"", `[[INT64(5)]]`)  // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a <= 5.00", `[[INT64(5)]]`)   // having clause
+
+			utils.AssertContainsError(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ 1 from aggr_test having count(*) <= 5", `expr cannot be converted, not supported: *sqlparser.FuncExpr`) // will fail since `count(*)` is a FuncExpr
+		})
+	}
+}
+
+func TestGreaterFilterOnScatter(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.Nil(t, err)
+	defer conn.Close()
+
+	utils.Exec(t, conn, "insert into aggr_test(id, val1, val2) values(1,'a',1), (2,'b',2), (3,'c',3), (4,'d',4), (5,'e',5)")
+
+	defer func() {
+		utils.Exec(t, conn, "set workload = oltp")
+		utils.Exec(t, conn, "delete from aggr_test")
+	}()
+
+	workloads := []string{"oltp", "olap"}
+	for _, workload := range workloads {
+		t.Run(workload, func(t *testing.T) {
+			utils.Exec(t, conn, fmt.Sprintf("set workload = '%s'", workload))
+
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a > 1", `[[INT64(5)]]`)     // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having 1 > a", `[]`)               // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a > a", `[]`)               // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a > 3+1", `[[INT64(5)]]`)   // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a > 10", `[]`)              // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a > \"1\"", `[[INT64(5)]]`) // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a > \"5\"", `[]`)           // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a > 4.00", `[[INT64(5)]]`)  // having clause
+
+			utils.AssertContainsError(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ 1 from aggr_test having count(*) > 5", `expr cannot be converted, not supported: *sqlparser.FuncExpr`) // will fail since `count(*)` is a FuncExpr
+		})
+	}
+}
+
+func TestGreaterEqualFilterOnScatter(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.Nil(t, err)
+	defer conn.Close()
+
+	utils.Exec(t, conn, "insert into aggr_test(id, val1, val2) values(1,'a',1), (2,'b',2), (3,'c',3), (4,'d',4), (5,'e',5)")
+
+	defer func() {
+		utils.Exec(t, conn, "set workload = oltp")
+		utils.Exec(t, conn, "delete from aggr_test")
+	}()
+
+	workloads := []string{"oltp", "olap"}
+	for _, workload := range workloads {
+		t.Run(workload, func(t *testing.T) {
+			utils.Exec(t, conn, fmt.Sprintf("set workload = '%s'", workload))
+
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a >= 1", `[[INT64(5)]]`)     // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having 1 >= a", `[]`)               // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a >= a", `[[INT64(5)]]`)     // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a >= 3+2", `[[INT64(5)]]`)   // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a >= 10", `[]`)              // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a >= \"1\"", `[[INT64(5)]]`) // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a >= \"5\"", `[[INT64(5)]]`) // having clause
+			utils.AssertMatches(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ count(*) as a from aggr_test having a >= 5.00", `[[INT64(5)]]`)  // having clause
+
+			utils.AssertContainsError(t, conn, "select /* GEN4_COMPARE_ONLY_GEN4 */ 1 from aggr_test having count(*) >= 5", `expr cannot be converted, not supported: *sqlparser.FuncExpr`) // will fail since `count(*)` is a FuncExpr
+		})
+	}
+}
diff --git a/go/test/endtoend/vtgate/queries/aggregation/main_test.go b/go/test/endtoend/vtgate/queries/aggregation/main_test.go
index efe5cc5bb8..82ed2c7762 100644
--- a/go/test/endtoend/vtgate/queries/aggregation/main_test.go
+++ b/go/test/endtoend/vtgate/queries/aggregation/main_test.go
@@ -51,6 +51,13 @@ create table aggr_test(
 	primary key(id)
 ) Engine=InnoDB;
 
+create table aggr_test_dates(
+	id bigint,
+	val1 datetime default current_timestamp,
+	val2 datetime default current_timestamp,
+	primary key(id)
+) Engine=InnoDB;
+
 create table t7_xxhash(
 	uid varchar(50),
 	phone bigint,
@@ -129,6 +136,24 @@ create table t7_xxhash_idx(
           "type": "VARCHAR"
         }
       ]
+    },
+	"aggr_test_dates": {
+      "column_vindexes": [
+        {
+          "column": "id",
+          "name": "hash"
+        }
+      ],
+      "columns": [
+        {
+          "name": "val1",
+          "type": "DATETIME"
+        },
+        {
+          "name": "val2",
+          "type": "DATETIME"
+        }
+      ]
     },
 	"t7_xxhash": {
       "column_vindexes": [
diff --git a/go/test/endtoend/vtgate/reservedconn/udv_test.go b/go/test/endtoend/vtgate/reservedconn/udv_test.go
index 391fc722a4..b548489c7f 100644
--- a/go/test/endtoend/vtgate/reservedconn/udv_test.go
+++ b/go/test/endtoend/vtgate/reservedconn/udv_test.go
@@ -100,7 +100,7 @@ func TestSetUDV(t *testing.T) {
 		expectedRows: `[]`,
 	}, {
 		query:        "select @foo = @bar, @dd, @tt",
-		expectedRows: `[[INT64(1) VARCHAR("2020-10-20") VARCHAR("10:15:00")]]`, rowsReturned: 1,
+		expectedRows: `[[INT32(1) DATE("2020-10-20") TIME("10:15:00")]]`, rowsReturned: 1,
 	}}
 
 	conn, err := mysql.Connect(ctx, &vtParams)
diff --git a/go/test/endtoend/vtgate/utils/utils.go b/go/test/endtoend/vtgate/utils/utils.go
index 1b3722daea..7f6a8dc0c2 100644
--- a/go/test/endtoend/vtgate/utils/utils.go
+++ b/go/test/endtoend/vtgate/utils/utils.go
@@ -39,6 +39,13 @@ func AssertMatches(t *testing.T, conn *mysql.Conn, query, expected string) {
 	}
 }
 
+func AssertContainsError(t *testing.T, conn *mysql.Conn, query, expected string) {
+	t.Helper()
+	_, err := ExecAllowError(t, conn, query)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), expected)
+}
+
 func AssertMatchesNoOrder(t *testing.T, conn *mysql.Conn, query, expected string) {
 	t.Helper()
 	qr := Exec(t, conn, query)
@@ -83,3 +90,8 @@ func Exec(t *testing.T, conn *mysql.Conn, query string) *sqltypes.Result {
 	require.NoError(t, err, "for query: "+query)
 	return qr
 }
+
+func ExecAllowError(t *testing.T, conn *mysql.Conn, query string) (*sqltypes.Result, error) {
+	t.Helper()
+	return conn.ExecuteFetch(query, 1000, true)
+}
diff --git a/go/vt/vtgate/engine/ordered_aggregate_test.go b/go/vt/vtgate/engine/ordered_aggregate_test.go
index 80fadaaf6b..2f6a27b582 100644
--- a/go/vt/vtgate/engine/ordered_aggregate_test.go
+++ b/go/vt/vtgate/engine/ordered_aggregate_test.go
@@ -448,6 +448,7 @@ func TestOrderedAggregateSumDistinctGood(t *testing.T) {
 			"c|4|1",
 			// Single null
 			"d|null|1",
+			"d|1|1",
 			// Start with null
 			"e|null|1",
 			"e|1|1",
@@ -496,14 +497,16 @@ func TestOrderedAggregateSumDistinctGood(t *testing.T) {
 		"a|1|3",
 		"b|1|1",
 		"c|7|2",
-		"d|null|1",
+		"d|1|2",
 		"e|1|2",
 		"f|1|2",
 		"g|6|4",
 		"h|6|4",
 		"i|7|2",
 	)
-	assert.Equal(wantResult, result)
+	want := fmt.Sprintf("%v", wantResult.Rows)
+	got := fmt.Sprintf("%v", result.Rows)
+	assert.Equal(want, got)
 }
 
 func TestOrderedAggregateSumDistinctTolerateError(t *testing.T) {
@@ -1073,7 +1076,7 @@ func TestOrderedAggregateCollate(t *testing.T) {
 		)},
 	}
 
-	collationID, _ := collations.IDFromName("utf8mb4_0900_ai_ci")
+	collationID, _ := collations.Default().LookupID("utf8mb4_0900_ai_ci")
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
 			Opcode: AggregateCount,
@@ -1115,7 +1118,7 @@ func TestOrderedAggregateCollateAS(t *testing.T) {
 		)},
 	}
 
-	collationID, _ := collations.IDFromName("utf8mb4_0900_as_ci")
+	collationID, _ := collations.Default().LookupID("utf8mb4_0900_as_ci")
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
 			Opcode: AggregateCount,
@@ -1159,7 +1162,7 @@ func TestOrderedAggregateCollateKS(t *testing.T) {
 		)},
 	}
 
-	collationID, _ := collations.IDFromName("utf8mb4_ja_0900_as_cs_ks")
+	collationID, _ := collations.Default().LookupID("utf8mb4_ja_0900_as_cs_ks")
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
 			Opcode: AggregateCount,
diff --git a/go/vt/vtgate/engine/route_test.go b/go/vt/vtgate/engine/route_test.go
index e2d8d5f218..6cf0b91f96 100644
--- a/go/vt/vtgate/engine/route_test.go
+++ b/go/vt/vtgate/engine/route_test.go
@@ -82,7 +82,7 @@ func TestSelectInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T)
 	stringListToExprList := func(in []string) []evalengine.Expr {
 		var schema []evalengine.Expr
 		for _, s := range in {
-			schema = append(schema, evalengine.NewLiteralString([]byte(s)))
+			schema = append(schema, evalengine.NewLiteralString([]byte(s), 0))
 		}
 		return schema
 	}
@@ -97,7 +97,7 @@ func TestSelectInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T)
 	tests := []testCase{{
 		testName:    "both schema and table predicates - routed table",
 		tableSchema: []string{"schema"},
-		tableName:   map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("table"))},
+		tableName:   map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("table"), 0)},
 		routed:      true,
 		expectedLog: []string{
 			"FindTable(`schema`.`table`)",
@@ -106,7 +106,7 @@ func TestSelectInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T)
 	}, {
 		testName:    "both schema and table predicates - not routed",
 		tableSchema: []string{"schema"},
-		tableName:   map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("table"))},
+		tableName:   map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("table"), 0)},
 		routed:      false,
 		expectedLog: []string{
 			"FindTable(`schema`.`table`)",
@@ -115,7 +115,7 @@ func TestSelectInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T)
 	}, {
 		testName:    "multiple schema and table predicates",
 		tableSchema: []string{"schema", "schema", "schema"},
-		tableName:   map[string]evalengine.Expr{"t1": evalengine.NewLiteralString([]byte("table")), "t2": evalengine.NewLiteralString([]byte("table")), "t3": evalengine.NewLiteralString([]byte("table"))},
+		tableName:   map[string]evalengine.Expr{"t1": evalengine.NewLiteralString([]byte("table"), 0), "t2": evalengine.NewLiteralString([]byte("table"), 0), "t3": evalengine.NewLiteralString([]byte("table"), 0)},
 		routed:      false,
 		expectedLog: []string{
 			"FindTable(`schema`.`table`)",
@@ -125,7 +125,7 @@ func TestSelectInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T)
 			"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"},
 	}, {
 		testName:  "table name predicate - routed table",
-		tableName: map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("tableName"))},
+		tableName: map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("tableName"), 0)},
 		routed:    true,
 		expectedLog: []string{
 			"FindTable(tableName)",
@@ -133,7 +133,7 @@ func TestSelectInformationSchemaWithTableAndSchemaWithRoutedTables(t *testing.T)
 			"ExecuteMultiShard routedKeyspace.1: dummy_select {table_name: type:VARBINARY value:\"routedTable\"} false false"},
 	}, {
 		testName:  "table name predicate - not routed",
-		tableName: map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("tableName"))},
+		tableName: map[string]evalengine.Expr{"table_name": evalengine.NewLiteralString([]byte("tableName"), 0)},
 		routed:    false,
 		expectedLog: []string{
 			"FindTable(tableName)",
diff --git a/go/vt/vtgate/engine/set_test.go b/go/vt/vtgate/engine/set_test.go
index 42700b6886..1464b0cee4 100644
--- a/go/vt/vtgate/engine/set_test.go
+++ b/go/vt/vtgate/engine/set_test.go
@@ -98,7 +98,7 @@ func TestSetTable(t *testing.T) {
 		setOps: []SetOp{
 			&UserDefinedVariable{
 				Name: "x",
-				Expr: evalengine.NewColumn(0),
+				Expr: evalengine.NewColumn(0, 0),
 			},
 		},
 		qr: []*sqltypes.Result{sqltypes.MakeTestResult(
diff --git a/go/vt/vtgate/evalengine/arithmetic_test.go b/go/vt/vtgate/evalengine/arithmetic_test.go
index dc57600703..f8cc3160e1 100644
--- a/go/vt/vtgate/evalengine/arithmetic_test.go
+++ b/go/vt/vtgate/evalengine/arithmetic_test.go
@@ -44,6 +44,8 @@ var (
 	NewUint64  = sqltypes.NewUint64
 	NewFloat64 = sqltypes.NewFloat64
 	TestValue  = sqltypes.TestValue
+
+	maxUint64 uint64 = math.MaxUint64
 )
 
 func TestArithmetics(t *testing.T) {
@@ -83,11 +85,11 @@ func TestArithmetics(t *testing.T) {
 			// testing for int64 overflow with min negative value
 			v1:  NewInt64(math.MinInt64),
 			v2:  NewInt64(1),
-			err: "BIGINT value is out of range in -9223372036854775808 - 1",
+			err: dataOutOfRangeError(math.MinInt64, 1, "BIGINT", "-").Error(),
 		}, {
 			v1:  NewUint64(4),
 			v2:  NewInt64(5),
-			err: "BIGINT UNSIGNED value is out of range in 4 - 5",
+			err: dataOutOfRangeError(4, 5, "BIGINT UNSIGNED", "-").Error(),
 		}, {
 			// testing uint - int
 			v1:  NewUint64(7),
@@ -101,7 +103,7 @@ func TestArithmetics(t *testing.T) {
 			// testing for int64 overflow
 			v1:  NewInt64(math.MinInt64),
 			v2:  NewUint64(0),
-			err: "BIGINT UNSIGNED value is out of range in -9223372036854775808 - 0",
+			err: dataOutOfRangeError(math.MinInt64, 0, "BIGINT UNSIGNED", "-").Error(),
 		}, {
 			v1:  TestValue(querypb.Type_VARCHAR, "c"),
 			v2:  NewInt64(1),
@@ -138,7 +140,7 @@ func TestArithmetics(t *testing.T) {
 		}, {
 			v1:  NewInt64(-1),
 			v2:  NewUint64(2),
-			err: "BIGINT UNSIGNED value is out of range in -1 - 2",
+			err: dataOutOfRangeError(-1, 2, "BIGINT UNSIGNED", "-").Error(),
 		}, {
 			v1:  NewInt64(2),
 			v2:  NewUint64(1),
@@ -167,7 +169,7 @@ func TestArithmetics(t *testing.T) {
 			// testing uint - uint if v2 > v1
 			v1:  NewUint64(2),
 			v2:  NewUint64(4),
-			err: "BIGINT UNSIGNED value is out of range in 2 - 4",
+			err: dataOutOfRangeError(2, 4, "BIGINT UNSIGNED", "-").Error(),
 		}, {
 			// testing uint - (- int)
 			v1:  NewUint64(1),
@@ -205,7 +207,7 @@ func TestArithmetics(t *testing.T) {
 		}, {
 			v1:  NewInt64(-2),
 			v2:  NewUint64(1),
-			err: "BIGINT UNSIGNED value is out of range in 1 + -2",
+			err: dataOutOfRangeError(1, -2, "BIGINT UNSIGNED", "+").Error(),
 		}, {
 			v1:  NewInt64(math.MaxInt64),
 			v2:  NewInt64(-2),
@@ -217,14 +219,14 @@ func TestArithmetics(t *testing.T) {
 			out: NewUint64(3),
 		}, {
 			// testing for overflow uint64
-			v1:  NewUint64(math.MaxUint64),
+			v1:  NewUint64(maxUint64),
 			v2:  NewUint64(2),
-			err: "BIGINT UNSIGNED value is out of range in 18446744073709551615 + 2",
+			err: dataOutOfRangeError(maxUint64, 2, "BIGINT UNSIGNED", "+").Error(),
 		}, {
 			// int64 underflow
 			v1:  NewInt64(math.MinInt64),
 			v2:  NewInt64(-2),
-			err: "BIGINT value is out of range in -9223372036854775808 + -2",
+			err: dataOutOfRangeError(math.MinInt64, -2, "BIGINT", "+").Error(),
 		}, {
 			// checking int64 max value can be returned
 			v1:  NewInt64(math.MaxInt64),
@@ -257,9 +259,9 @@ func TestArithmetics(t *testing.T) {
 			err: "strconv.ParseInt: parsing \"1.2\": invalid syntax",
 		}, {
 			// testing for uint64 overflow with max uint64 + int value
-			v1:  NewUint64(math.MaxUint64),
+			v1:  NewUint64(maxUint64),
 			v2:  NewInt64(2),
-			err: "BIGINT UNSIGNED value is out of range in 18446744073709551615 + 2",
+			err: dataOutOfRangeError(maxUint64, 2, "BIGINT UNSIGNED", "+").Error(),
 		}},
 	}, {
 		operator: "/",
@@ -343,7 +345,7 @@ func TestArithmetics(t *testing.T) {
 			// testing for overflow of float64
 			v1:  NewFloat64(math.MaxFloat64),
 			v2:  NewFloat64(0.5),
-			err: "BIGINT value is out of range in 1.7976931348623157e+308 / 0.5",
+			err: dataOutOfRangeError(math.MaxFloat64, 0.5, "BIGINT", "/").Error(),
 		}},
 	}, {
 		operator: "*",
@@ -407,12 +409,12 @@ func TestArithmetics(t *testing.T) {
 			// testing for overflow of int64
 			v1:  NewInt64(math.MaxInt64),
 			v2:  NewInt64(2),
-			err: "BIGINT value is out of range in 9223372036854775807 * 2",
+			err: dataOutOfRangeError(math.MaxInt64, 2, "BIGINT", "*").Error(),
 		}, {
 			// testing for underflow of uint64*max.uint64
 			v1:  NewInt64(2),
-			v2:  NewUint64(math.MaxUint64),
-			err: "BIGINT UNSIGNED value is out of range in 18446744073709551615 * 2",
+			v2:  NewUint64(maxUint64),
+			err: dataOutOfRangeError(maxUint64, 2, "BIGINT UNSIGNED", "*").Error(),
 		}, {
 			v1:  NewUint64(math.MaxUint64),
 			v2:  NewUint64(1),
@@ -421,7 +423,7 @@ func TestArithmetics(t *testing.T) {
 			//Checking whether maxInt value can be passed as uint value
 			v1:  NewUint64(math.MaxInt64),
 			v2:  NewInt64(3),
-			err: "BIGINT UNSIGNED value is out of range in 9223372036854775807 * 3",
+			err: dataOutOfRangeError(math.MaxInt64, 3, "BIGINT UNSIGNED", "*").Error(),
 		}},
 	}}
 
@@ -443,7 +445,7 @@ func TestArithmetics(t *testing.T) {
 	}
 }
 
-func TestNullsafeAdd(t *testing.T) {
+func TestNullSafeAdd(t *testing.T) {
 	tcases := []struct {
 		v1, v2 sqltypes.Value
 		out    sqltypes.Value
@@ -482,7 +484,11 @@ func TestNullsafeAdd(t *testing.T) {
 		// Make sure underlying error is returned while adding.
 		v1:  NewInt64(-1),
 		v2:  NewUint64(2),
-		out: NewInt64(-9223372036854775808),
+		out: NewInt64(1),
+	}, {
+		v1:  NewInt64(-100),
+		v2:  NewUint64(10),
+		err: dataOutOfRangeError(10, -100, "BIGINT UNSIGNED", "+"),
 	}, {
 		// Make sure underlying error is returned while converting.
 		v1:  NewFloat64(1),
@@ -490,10 +496,16 @@ func TestNullsafeAdd(t *testing.T) {
 		out: NewInt64(3),
 	}}
 	for _, tcase := range tcases {
-		got := NullsafeAdd(tcase.v1, tcase.v2, querypb.Type_INT64)
+		got, err := NullSafeAdd(tcase.v1, tcase.v2, querypb.Type_INT64)
+
+		if tcase.err == nil {
+			require.NoError(t, err)
+		} else {
+			require.EqualError(t, err, tcase.err.Error())
+		}
 
 		if !reflect.DeepEqual(got, tcase.out) {
-			t.Errorf("NullsafeAdd(%v, %v): %v, want %v", printValue(tcase.v1), printValue(tcase.v2), printValue(got), printValue(tcase.out))
+			t.Errorf("NullSafeAdd(%v, %v): %v, want %v", printValue(tcase.v1), printValue(tcase.v2), printValue(got), printValue(tcase.out))
 		}
 	}
 }
@@ -600,7 +612,7 @@ func TestNullsafeCompare(t *testing.T) {
 }
 
 func getCollationID(collation string) collations.ID {
-	id, _ := collations.IDFromName(collation)
+	id, _ := collations.Default().LookupID(collation)
 	return id
 }
 
@@ -1202,9 +1214,9 @@ func TestPrioritize(t *testing.T) {
 	}}
 	for _, tcase := range tcases {
 		t.Run(tcase.v1.Value().String()+" - "+tcase.v2.Value().String(), func(t *testing.T) {
-			got1, got2 := makeNumericAndprioritize(tcase.v1, tcase.v2)
-			utils.MustMatch(t, tcase.out1, got1, "makeNumericAndprioritize")
-			utils.MustMatch(t, tcase.out2, got2, "makeNumericAndprioritize")
+			got1, got2 := makeNumericAndPrioritize(tcase.v1, tcase.v2)
+			utils.MustMatch(t, tcase.out1, got1, "makeNumericAndPrioritize")
+			utils.MustMatch(t, tcase.out2, got2, "makeNumericAndPrioritize")
 		})
 	}
 }
@@ -1581,7 +1593,7 @@ func BenchmarkAddActual(b *testing.B) {
 	v1 := sqltypes.MakeTrusted(querypb.Type_INT64, []byte("1"))
 	v2 := sqltypes.MakeTrusted(querypb.Type_INT64, []byte("12"))
 	for i := 0; i < b.N; i++ {
-		v1 = NullsafeAdd(v1, v2, querypb.Type_INT64)
+		v1, _ = NullSafeAdd(v1, v2, querypb.Type_INT64)
 	}
 }
 
diff --git a/go/vt/vtgate/evalengine/expressions_test.go b/go/vt/vtgate/evalengine/binary_test.go
similarity index 99%
rename from go/vt/vtgate/evalengine/expressions_test.go
rename to go/vt/vtgate/evalengine/binary_test.go
index 0cfda38dde..514aa43882 100644
--- a/go/vt/vtgate/evalengine/expressions_test.go
+++ b/go/vt/vtgate/evalengine/binary_test.go
@@ -35,7 +35,7 @@ func TestBinaryOpTypes(t *testing.T) {
 		l, r, e querypb.Type
 	}
 	type ops struct {
-		op        BinaryExpr
+		op        BinaryOp
 		testcases []testcase
 	}
 
diff --git a/go/vt/vtgate/evalengine/comparisons_test.go b/go/vt/vtgate/evalengine/comparisons_test.go
new file mode 100644
index 0000000000..a576515e73
--- /dev/null
+++ b/go/vt/vtgate/evalengine/comparisons_test.go
@@ -0,0 +1,896 @@
+/*
+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 (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/mysql/collations"
+
+	"vitess.io/vitess/go/sqltypes"
+
+	querypb "vitess.io/vitess/go/vt/proto/query"
+)
+
+type testCase struct {
+	name   string
+	v1, v2 Expr
+	out    *bool
+	err    string
+	op     ComparisonOp
+	bv     map[string]*querypb.BindVariable
+	row    []sqltypes.Value
+}
+
+var (
+	T = true
+	F = false
+
+	defaultCollation = collations.Default().LookupByName("utf8mb4_bin").ID()
+)
+
+func (tc testCase) run(t *testing.T) {
+	if tc.bv == nil {
+		tc.bv = map[string]*querypb.BindVariable{}
+	}
+	env := ExpressionEnv{
+		BindVars: tc.bv,
+		Row:      tc.row,
+	}
+	cmp := ComparisonExpr{
+		Op:    tc.op,
+		Left:  tc.v1,
+		Right: tc.v2,
+	}
+	got, err := cmp.Evaluate(env)
+	if tc.err == "" {
+		require.NoError(t, err)
+		if tc.out != nil && *tc.out {
+			require.EqualValues(t, 1, got.ival)
+		} else if tc.out != nil && !*tc.out {
+			require.EqualValues(t, 0, got.ival)
+		} else {
+			require.EqualValues(t, 0, got.ival)
+			require.EqualValues(t, sqltypes.Null, got.typ)
+		}
+	} else {
+		require.EqualError(t, err, tc.err)
+	}
+}
+
+// This test tests the comparison of two integers
+func TestCompareIntegers(t *testing.T) {
+	tests := []testCase{
+		{
+			name: "integers are equal (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(0, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewInt64(18)},
+		},
+		{
+			name: "integers are equal (2)",
+			v1:   NewLiteralInt(56), v2: NewLiteralInt(56),
+			out: &T, op: &EqualOp{},
+		},
+		{
+			name: "integers are not equal (1)",
+			v1:   NewLiteralInt(56), v2: NewLiteralInt(10),
+			out: &F, op: &EqualOp{},
+		},
+		{
+			name: "integers are not equal (2)",
+			v1:   NewLiteralInt(56), v2: NewLiteralInt(10),
+			out: &T, op: &NotEqualOp{},
+		},
+		{
+			name: "integers are not equal (3)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewInt64(18), sqltypes.NewInt64(98)},
+		},
+		{
+			name: "unsigned integers are equal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(0, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewUint64(18)},
+		},
+		{
+			name: "unsigned integer and integer are equal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewUint64(18), sqltypes.NewInt64(18)},
+		},
+		{
+			name: "unsigned integer and integer are not equal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewUint64(18), sqltypes.NewInt64(42)},
+		},
+		{
+			name: "integer is less than integer",
+			v1:   NewLiteralInt(3549), v2: NewLiteralInt(8072),
+			out: &T, op: &LessThanOp{},
+		},
+		{
+			name: "integer is not less than integer",
+			v1:   NewLiteralInt(3549), v2: NewLiteralInt(21),
+			out: &F, op: &LessThanOp{},
+		},
+		{
+			name: "integer is less-equal to integer (1)",
+			v1:   NewLiteralInt(3549), v2: NewLiteralInt(9863),
+			out: &T, op: &LessEqualOp{},
+		},
+		{
+			name: "integer is less-equal to integer (2)",
+			v1:   NewLiteralInt(3549), v2: NewLiteralInt(3549),
+			out: &T, op: &LessEqualOp{},
+		},
+		{
+			name: "integer is greater than integer",
+			v1:   NewLiteralInt(9809), v2: NewLiteralInt(9800),
+			out: &T, op: &GreaterThanOp{},
+		},
+		{
+			name: "integer is not greater than integer",
+			v1:   NewLiteralInt(549), v2: NewLiteralInt(21579),
+			out: &F, op: &GreaterThanOp{},
+		},
+		{
+			name: "integer is greater-equal to integer (1)",
+			v1:   NewLiteralInt(987), v2: NewLiteralInt(15),
+			out: &T, op: &GreaterEqualOp{},
+		},
+		{
+			name: "integer is greater-equal to integer (2)",
+			v1:   NewLiteralInt(3549), v2: NewLiteralInt(3549),
+			out: &T, op: &GreaterEqualOp{},
+		},
+	}
+
+	for i, tcase := range tests {
+		t.Run(fmt.Sprintf("%d %s", i, tcase.name), func(t *testing.T) {
+			tcase.run(t)
+		})
+	}
+}
+
+// This test tests the comparison of two floats
+func TestCompareFloats(t *testing.T) {
+	tests := []testCase{
+		{
+			name: "floats are equal (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(0, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewFloat64(18)},
+		},
+		{
+			name: "floats are equal (2)",
+			v1:   NewLiteralFloat(3549.9), v2: NewLiteralFloat(3549.9),
+			out: &T, op: &EqualOp{},
+		},
+		{
+			name: "floats are not equal (1)",
+			v1:   NewLiteralFloat(7858.016), v2: NewLiteralFloat(8943298.56),
+			out: &F, op: &EqualOp{},
+		},
+		{
+			name: "floats are not equal (2)",
+			v1:   NewLiteralFloat(351049.65), v2: NewLiteralFloat(62508.99),
+			out: &T, op: &NotEqualOp{},
+		},
+		{
+			name: "floats are not equal (3)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewFloat64(16516.84), sqltypes.NewFloat64(219541.01)},
+		},
+		{
+			name: "float is less than float",
+			v1:   NewLiteralFloat(3549.9), v2: NewLiteralFloat(8072),
+			out: &T, op: &LessThanOp{},
+		},
+		{
+			name: "float is not less than float",
+			v1:   NewLiteralFloat(3549.9), v2: NewLiteralFloat(21.564),
+			out: &F, op: &LessThanOp{},
+		},
+		{
+			name: "float is less-equal to float (1)",
+			v1:   NewLiteralFloat(3549.9), v2: NewLiteralFloat(9863),
+			out: &T, op: &LessEqualOp{},
+		},
+		{
+			name: "float is less-equal to float (2)",
+			v1:   NewLiteralFloat(3549.9), v2: NewLiteralFloat(3549.9),
+			out: &T, op: &LessEqualOp{},
+		},
+		{
+			name: "float is greater than float",
+			v1:   NewLiteralFloat(9808.549), v2: NewLiteralFloat(9808.540),
+			out: &T, op: &GreaterThanOp{},
+		},
+		{
+			name: "float is not greater than float",
+			v1:   NewLiteralFloat(549.02), v2: NewLiteralFloat(21579.64),
+			out: &F, op: &GreaterThanOp{},
+		},
+		{
+			name: "float is greater-equal to float (1)",
+			v1:   NewLiteralFloat(987.30), v2: NewLiteralFloat(15.5),
+			out: &T, op: &GreaterEqualOp{},
+		},
+		{
+			name: "float is greater-equal to float (2)",
+			v1:   NewLiteralFloat(3549.9), v2: NewLiteralFloat(3549.9),
+			out: &T, op: &GreaterEqualOp{},
+		},
+	}
+
+	for i, tcase := range tests {
+		t.Run(fmt.Sprintf("%d %s", i, tcase.name), func(t *testing.T) {
+			tcase.run(t)
+		})
+	}
+}
+
+// This test tests the comparison of two decimals
+func TestCompareDecimals(t *testing.T) {
+	tests := []testCase{
+		{
+			name: "decimals are equal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(0, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("12.9019")},
+		},
+		{
+			name: "decimals are not equal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("12.9019"), sqltypes.NewDecimal("489.156849")},
+		},
+		{
+			name: "decimal is greater than decimal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("192.129"), sqltypes.NewDecimal("192.128")},
+		},
+		{
+			name: "decimal is not greater than decimal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("192.128"), sqltypes.NewDecimal("192.129")},
+		},
+		{
+			name: "decimal is less than decimal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("192.128"), sqltypes.NewDecimal("192.129")},
+		},
+		{
+			name: "decimal is not less than decimal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &LessThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("192.129"), sqltypes.NewDecimal("192.128")},
+		},
+	}
+
+	for i, tcase := range tests {
+		t.Run(fmt.Sprintf("%d %s", i, tcase.name), func(t *testing.T) {
+			tcase.run(t)
+		})
+	}
+}
+
+// This test tests the comparison of numerical values (float, decimal, integer)
+func TestCompareNumerics(t *testing.T) {
+	tests := []testCase{
+		{
+			name: "decimal and float are equal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewFloat64(189.6), sqltypes.NewDecimal("189.6")},
+		},
+		{
+			name: "decimal and float with negative values are equal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewFloat64(-98.1839), sqltypes.NewDecimal("-98.1839")},
+		},
+		{
+			name: "decimal and float with negative values are not equal (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewFloat64(-98.9381), sqltypes.NewDecimal("-98.1839")},
+		},
+		{
+			name: "decimal and float with negative values are not equal (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewFloat64(-98.9381), sqltypes.NewDecimal("-98.1839")},
+		},
+		{
+			name: "decimal and integer are equal (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewInt64(8979), sqltypes.NewDecimal("8979")},
+		},
+		{
+			name: "decimal and integer are equal (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("8979.0000"), sqltypes.NewInt64(8979)},
+		},
+		{
+			name: "decimal and unsigned integer are equal (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewUint64(901), sqltypes.NewDecimal("901")},
+		},
+		{
+			name: "decimal and unsigned integer are equal (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("901.00"), sqltypes.NewUint64(901)},
+		},
+		{
+			name: "decimal and unsigned integer are not equal (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("192.129"), sqltypes.NewUint64(192)},
+		},
+		{
+			name: "decimal and unsigned integer are not equal (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("192.129"), sqltypes.NewUint64(192)},
+		},
+		{
+			name: "decimal is greater than integer",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("1.01"), sqltypes.NewInt64(1)},
+		},
+		{
+			name: "decimal is greater-equal to integer",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("1.00"), sqltypes.NewInt64(1)},
+		},
+		{
+			name: "decimal is less than integer",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal(".99"), sqltypes.NewInt64(1)},
+		},
+		{
+			name: "decimal is less-equal to integer",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("1.00"), sqltypes.NewInt64(1)},
+		},
+		{
+			name: "decimal is greater than float",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("849.896"), sqltypes.NewFloat64(86.568)},
+		},
+		{
+			name: "decimal is not greater than float",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("15.23"), sqltypes.NewFloat64(8689.5)},
+		},
+		{
+			name: "decimal is greater-equal to float (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("65"), sqltypes.NewFloat64(65)},
+		},
+		{
+			name: "decimal is greater-equal to float (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("65"), sqltypes.NewFloat64(60)},
+		},
+		{
+			name: "decimal is less than float",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("0.998"), sqltypes.NewFloat64(0.999)},
+		},
+		{
+			name: "decimal is less-equal to float",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDecimal("1.000101"), sqltypes.NewFloat64(1.00101)},
+		},
+	}
+
+	for i, tcase := range tests {
+		t.Run(fmt.Sprintf("%d %s", i, tcase.name), func(t *testing.T) {
+			tcase.run(t)
+		})
+	}
+}
+
+// This test tests the comparison of two datetimes
+func TestCompareDatetime(t *testing.T) {
+	tests := []testCase{
+		{
+			name: "datetimes are equal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(0, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDatetime("2021-10-22 12:00:00")},
+		},
+		{
+			name: "datetimes are not equal (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDatetime("2021-10-22 12:00:00"), sqltypes.NewDatetime("2020-10-22 12:00:00")},
+		},
+		{
+			name: "datetimes are not equal (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDatetime("2021-10-22 12:00:00"), sqltypes.NewDatetime("2021-10-22 10:23:56")},
+		},
+		{
+			name: "datetimes are not equal (3)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDatetime("2021-10-01 00:00:00"), sqltypes.NewDatetime("2021-02-01 00:00:00")},
+		},
+		{
+			name: "datetime is greater than datetime",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDatetime("2021-10-30 10:42:50"), sqltypes.NewDatetime("2021-10-01 13:10:02")},
+		},
+		{
+			name: "datetime is not greater than datetime",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDatetime("2021-10-01 13:10:02"), sqltypes.NewDatetime("2021-10-30 10:42:50")},
+		},
+		{
+			name: "datetime is less than datetime",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDatetime("2021-10-01 13:10:02"), sqltypes.NewDatetime("2021-10-30 10:42:50")},
+		},
+		{
+			name: "datetime is not less than datetime",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &LessThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDatetime("2021-10-30 10:42:50"), sqltypes.NewDatetime("2021-10-01 13:10:02")},
+		},
+		{
+			name: "datetime is greater-equal to datetime (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDatetime("2021-10-30 10:42:50"), sqltypes.NewDatetime("2021-10-30 10:42:50")},
+		},
+		{
+			name: "datetime is greater-equal to datetime (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDatetime("2021-10-30 10:42:50"), sqltypes.NewDatetime("2021-10-01 13:10:02")},
+		},
+		{
+			name: "datetime is less-equal to datetime (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDatetime("2021-10-30 10:42:50"), sqltypes.NewDatetime("2021-10-30 10:42:50")},
+		},
+		{
+			name: "datetime is less-equal to datetime (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDatetime("2021-10-01 13:10:02"), sqltypes.NewDatetime("2021-10-30 10:42:50")},
+		},
+	}
+
+	for i, tcase := range tests {
+		t.Run(fmt.Sprintf("%d %s", i, tcase.name), func(t *testing.T) {
+			tcase.run(t)
+		})
+	}
+}
+
+// This test tests the comparison of two timestamps
+func TestCompareTimestamp(t *testing.T) {
+	tests := []testCase{
+		{
+			name: "timestamps are equal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(0, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTimestamp("2021-10-22 12:00:00")},
+		},
+		{
+			name: "timestamps are not equal (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTimestamp("2021-10-22 12:00:00"), sqltypes.NewTimestamp("2020-10-22 12:00:00")},
+		},
+		{
+			name: "timestamps are not equal (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTimestamp("2021-10-22 12:00:00"), sqltypes.NewTimestamp("2021-10-22 10:23:56")},
+		},
+		{
+			name: "timestamps are not equal (3)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTimestamp("2021-10-01 00:00:00"), sqltypes.NewTimestamp("2021-02-01 00:00:00")},
+		},
+		{
+			name: "timestamp is greater than timestamp",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewTimestamp("2021-10-30 10:42:50"), sqltypes.NewTimestamp("2021-10-01 13:10:02")},
+		},
+		{
+			name: "timestamp is not greater than timestamp",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewTimestamp("2021-10-01 13:10:02"), sqltypes.NewTimestamp("2021-10-30 10:42:50")},
+		},
+		{
+			name: "timestamp is less than timestamp",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessThanOp{},
+			row: []sqltypes.Value{sqltypes.NewTimestamp("2021-10-01 13:10:02"), sqltypes.NewTimestamp("2021-10-30 10:42:50")},
+		},
+		{
+			name: "timestamp is not less than timestamp",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &LessThanOp{},
+			row: []sqltypes.Value{sqltypes.NewTimestamp("2021-10-30 10:42:50"), sqltypes.NewTimestamp("2021-10-01 13:10:02")},
+		},
+		{
+			name: "timestamp is greater-equal to timestamp (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTimestamp("2021-10-30 10:42:50"), sqltypes.NewTimestamp("2021-10-30 10:42:50")},
+		},
+		{
+			name: "timestamp is greater-equal to timestamp (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTimestamp("2021-10-30 10:42:50"), sqltypes.NewTimestamp("2021-10-01 13:10:02")},
+		},
+		{
+			name: "timestamp is less-equal to timestamp (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTimestamp("2021-10-30 10:42:50"), sqltypes.NewTimestamp("2021-10-30 10:42:50")},
+		},
+		{
+			name: "timestamp is less-equal to timestamp (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTimestamp("2021-10-01 13:10:02"), sqltypes.NewTimestamp("2021-10-30 10:42:50")},
+		},
+	}
+
+	for i, tcase := range tests {
+		t.Run(fmt.Sprintf("%d %s", i, tcase.name), func(t *testing.T) {
+			tcase.run(t)
+		})
+	}
+}
+
+// This test tests the comparison of two dates
+func TestCompareDate(t *testing.T) {
+	tests := []testCase{
+		{
+			name: "dates are equal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(0, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-22")},
+		},
+		{
+			name: "dates are not equal (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-22"), sqltypes.NewDate("2020-10-21")},
+		},
+		{
+			name: "dates are not equal (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-01"), sqltypes.NewDate("2021-02-01")},
+		},
+		{
+			name: "date is greater than date",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-30"), sqltypes.NewDate("2021-10-01")},
+		},
+		{
+			name: "date is not greater than date",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-01"), sqltypes.NewDate("2021-10-30")},
+		},
+		{
+			name: "date is less than date",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-01"), sqltypes.NewDate("2021-10-30")},
+		},
+		{
+			name: "date is not less than date",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &LessThanOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-30"), sqltypes.NewDate("2021-10-01")},
+		},
+		{
+			name: "date is greater-equal to date (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-30"), sqltypes.NewDate("2021-10-30")},
+		},
+		{
+			name: "date is greater-equal to date (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-30"), sqltypes.NewDate("2021-10-01")},
+		},
+		{
+			name: "date is less-equal to date (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-30"), sqltypes.NewDate("2021-10-30")},
+		},
+		{
+			name: "date is less-equal to date (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-01"), sqltypes.NewDate("2021-10-30")},
+		},
+	}
+
+	for i, tcase := range tests {
+		t.Run(fmt.Sprintf("%d %s", i, tcase.name), func(t *testing.T) {
+			tcase.run(t)
+		})
+	}
+}
+
+// This test tests the comparison of two times
+func TestCompareTime(t *testing.T) {
+	tests := []testCase{
+		{
+			name: "times are equal",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(0, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTime("12:00:00")},
+		},
+		{
+			name: "times are not equal (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTime("12:00:00"), sqltypes.NewTime("10:23:56")},
+		},
+		{
+			name: "times are not equal (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTime("00:00:00"), sqltypes.NewTime("10:15:00")},
+		},
+		{
+			name: "time is greater than time",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewTime("18:14:35"), sqltypes.NewTime("13:01:38")},
+		},
+		{
+			name: "time is not greater than time",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &GreaterThanOp{},
+			row: []sqltypes.Value{sqltypes.NewTime("02:46:02"), sqltypes.NewTime("10:42:50")},
+		},
+		{
+			name: "time is less than time",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessThanOp{},
+			row: []sqltypes.Value{sqltypes.NewTime("04:30:00"), sqltypes.NewTime("09:23:48")},
+		},
+		{
+			name: "time is not less than time",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &F, op: &LessThanOp{},
+			row: []sqltypes.Value{sqltypes.NewTime("15:21:00"), sqltypes.NewTime("10:00:00")},
+		},
+		{
+			name: "time is greater-equal to time (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTime("10:42:50"), sqltypes.NewTime("10:42:50")},
+		},
+		{
+			name: "time is greater-equal to time (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &GreaterEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTime("19:42:50"), sqltypes.NewTime("13:10:02")},
+		},
+		{
+			name: "time is less-equal to time (1)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTime("10:42:50"), sqltypes.NewTime("10:42:50")},
+		},
+		{
+			name: "time is less-equal to time (2)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &LessEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewTime("10:10:02"), sqltypes.NewTime("10:42:50")},
+		},
+	}
+
+	for i, tcase := range tests {
+		t.Run(fmt.Sprintf("%d %s", i, tcase.name), func(t *testing.T) {
+			tcase.run(t)
+		})
+	}
+}
+
+// This test tests the comparison of two dates (datetime, date, timestamp, time)
+func TestCompareDates(t *testing.T) {
+	tests := []testCase{
+		{
+			name: "date equal datetime",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-22"), sqltypes.NewDatetime("2021-10-22 00:00:00")},
+		},
+		{
+			name: "date equal datetime through bind variables",
+			v1:   NewBindVar("k1", defaultCollation), v2: NewBindVar("k2", defaultCollation),
+			out: &T, op: &EqualOp{},
+			bv: map[string]*querypb.BindVariable{
+				"k1": {Type: sqltypes.Date, Value: []byte("2021-10-22")},
+				"k2": {Type: sqltypes.Datetime, Value: []byte("2021-10-22 00:00:00")},
+			},
+		},
+		{
+			name: "date not equal datetime through bind variables",
+			v1:   NewBindVar("k1", defaultCollation), v2: NewBindVar("k2", defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			bv: map[string]*querypb.BindVariable{
+				"k1": {Type: sqltypes.Date, Value: []byte("2021-02-20")},
+				"k2": {Type: sqltypes.Datetime, Value: []byte("2021-10-22 00:00:00")},
+			},
+		},
+		{
+			name: "date not equal datetime",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-22"), sqltypes.NewDatetime("2021-10-20 00:06:00")},
+		},
+		{
+			name: "date equal timestamp",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-22"), sqltypes.NewTimestamp("2021-10-22 00:00:00")},
+		},
+		{
+			name: "date not equal timestamp",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-10-22"), sqltypes.NewTimestamp("2021-10-22 16:00:00")},
+		},
+		{
+			name: "date equal time",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate(time.Now().Format("2006-01-02")), sqltypes.NewTime("00:00:00")},
+		},
+		{
+			name: "date not equal time",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate(time.Now().Format("2006-01-02")), sqltypes.NewTime("12:00:00")},
+		},
+		{
+			name: "string equal datetime",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewVarChar("2021-10-22"), sqltypes.NewDatetime("2021-10-22 00:00:00")},
+		},
+		{
+			name: "string equal timestamp",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewVarChar("2021-10-22 00:00:00"), sqltypes.NewTimestamp("2021-10-22 00:00:00")},
+		},
+		{
+			name: "string not equal timestamp",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewVarChar("2021-10-22 06:00:30"), sqltypes.NewTimestamp("2021-10-20 15:02:10")},
+		},
+		{
+			name: "string equal time",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewVarChar("00:05:12"), sqltypes.NewTime("00:05:12")},
+		},
+		{
+			name: "string equal date",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewVarChar("2021-02-22"), sqltypes.NewDate("2021-02-22")},
+		},
+		{
+			name: "string not equal date (1, date on the RHS)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewVarChar("2021-02-20"), sqltypes.NewDate("2021-03-30")},
+		},
+		{
+			name: "string not equal date (2, date on the LHS)",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &NotEqualOp{},
+			row: []sqltypes.Value{sqltypes.NewDate("2021-03-30"), sqltypes.NewVarChar("2021-02-20")},
+		},
+	}
+
+	for i, tcase := range tests {
+		t.Run(fmt.Sprintf("%d %s", i, tcase.name), func(t *testing.T) {
+			tcase.run(t)
+		})
+	}
+}
+
+// This test tests the comparison of strings
+func TestCompareStrings(t *testing.T) {
+	tests := []testCase{
+		{
+			name: "string equal string",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewVarChar("toto"), sqltypes.NewVarChar("toto")},
+		},
+		{
+			name: "string equal number",
+			v1:   NewColumn(0, defaultCollation), v2: NewColumn(1, defaultCollation),
+			out: &T, op: &EqualOp{},
+			row: []sqltypes.Value{sqltypes.NewVarChar("1"), sqltypes.NewInt64(1)},
+		},
+		{
+			name: "string equal string unknown collation",
+			v1:   NewColumn(0, collations.Unknown), v2: NewColumn(1, collations.Unknown),
+			op:  &EqualOp{},
+			err: "cannot compare strings with an unknown collation",
+			row: []sqltypes.Value{sqltypes.NewVarChar("1"), sqltypes.NewVarChar("1")},
+		},
+	}
+
+	for i, tcase := range tests {
+		t.Run(fmt.Sprintf("%d %s", i, tcase.name), func(t *testing.T) {
+			tcase.run(t)
+		})
+	}
+}
diff --git a/go/vt/sqlparser/expressions_test.go b/go/vt/vtgate/evalengine/convert_test.go
similarity index 88%
rename from go/vt/sqlparser/expressions_test.go
rename to go/vt/vtgate/evalengine/convert_test.go
index a72c50f198..740707ecd0 100644
--- a/go/vt/sqlparser/expressions_test.go
+++ b/go/vt/vtgate/evalengine/convert_test.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2020 The Vitess Authors.
+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.
@@ -14,12 +14,12 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package sqlparser
+package evalengine
 
 import (
 	"testing"
 
-	"vitess.io/vitess/go/vt/vtgate/evalengine"
+	"vitess.io/vitess/go/vt/sqlparser"
 
 	"vitess.io/vitess/go/sqltypes"
 
@@ -75,13 +75,13 @@ func TestEvaluate(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.expression, func(t *testing.T) {
 			// Given
-			stmt, err := Parse("select " + test.expression)
+			stmt, err := sqlparser.Parse("select " + test.expression)
 			require.NoError(t, err)
-			astExpr := stmt.(*Select).SelectExprs[0].(*AliasedExpr).Expr
-			sqltypesExpr, err := Convert(astExpr)
+			astExpr := stmt.(*sqlparser.Select).SelectExprs[0].(*sqlparser.AliasedExpr).Expr
+			sqltypesExpr, err := Convert(astExpr, nil)
 			require.Nil(t, err)
 			require.NotNil(t, sqltypesExpr)
-			env := evalengine.ExpressionEnv{
+			env := ExpressionEnv{
 				BindVars: map[string]*querypb.BindVariable{
 					"exp":                  sqltypes.Int64BindVariable(66),
 					"string_bind_variable": sqltypes.StringBindVariable("bar"),
diff --git a/go/vt/vtgate/planbuilder/expression_converter_test.go b/go/vt/vtgate/planbuilder/expression_converter_test.go
index 2ff05d188f..86ad0061c2 100644
--- a/go/vt/vtgate/planbuilder/expression_converter_test.go
+++ b/go/vt/vtgate/planbuilder/expression_converter_test.go
@@ -40,7 +40,7 @@ func TestConversion(t *testing.T) {
 		expressionsOut: e(evalengine.NewLiteralInt(1)),
 	}, {
 		expressionsIn:  "@@foo",
-		expressionsOut: e(evalengine.NewColumn(0)),
+		expressionsOut: e(evalengine.NewColumn(0, 0)),
 	}}
 
 	for _, tc := range queries {
diff --git a/go/vt/vtgate/planbuilder/rewrite_test.go b/go/vt/vtgate/planbuilder/rewrite_test.go
index f774fa3d56..d4815deebb 100644
--- a/go/vt/vtgate/planbuilder/rewrite_test.go
+++ b/go/vt/vtgate/planbuilder/rewrite_test.go
@@ -120,7 +120,10 @@ func TestHavingRewrite(t *testing.T) {
 		output: "select 1 from t1 where (x = 1 or y = 2) and a = 1 having count(*) = 1",
 	}, {
 		input:  "select count(*) k from t1 where x = 1 or y = 2 having a = 1 and k = 1",
-		output: "select count(*) as k from t1 where (x = 1 or y = 2) and a = 1 having count(*) = 1",
+		output: "select count(*) as k from t1 where (x = 1 or y = 2) and a = 1 having k = 1",
+	}, {
+		input:  "select count(*) k from t1 having k = 10",
+		output: "select count(*) as k from t1 having k = 10",
 	}, {
 		input:  "select 1 from t1 where x in (select 1 from t2 having a = 1)",
 		output: "select 1 from t1 where :__sq_has_values1 = 1 and x in ::__sq1",
diff --git a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
index e7e8bc2a7a..777d769801 100644
--- a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
@@ -808,7 +808,7 @@ Gen4 error: Expression of SELECT list is not in GROUP BY clause and contains non
       "Sharded": true
     },
     "FieldQuery": "select id, count(*) as c from `user` where 1 != 1 group by id",
-    "Query": "select id, count(*) as c from `user` where id = 1 group by id having count(*) = 10",
+    "Query": "select id, count(*) as c from `user` where id = 1 group by id having c = 10",
     "Table": "`user`",
     "Values": [
       1
@@ -2825,3 +2825,263 @@ Gen4 plan same as above
   }
 }
 
+# Column and Literal equality filter on scatter aggregates
+"select count(*) a from user having a = 10"
+"unsupported: filtering on results of aggregates"
+{
+  "QueryType": "SELECT",
+  "Original": "select count(*) a from user having a = 10",
+  "Instructions": {
+    "OperatorType": "Filter",
+    "Predicate": "a = 10",
+    "Inputs": [
+      {
+        "OperatorType": "Aggregate",
+        "Variant": "Ordered",
+        "Aggregates": "count(0) AS a",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "SelectScatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select count(*) as a from `user` where 1 != 1",
+            "Query": "select count(*) as a from `user`",
+            "Table": "`user`"
+          }
+        ]
+      }
+    ]
+  }
+}
+
+# Equality filtering with column and string literal on scatter aggregates
+"select count(*) a from user having a = '1'"
+"unsupported: filtering on results of aggregates"
+{
+  "QueryType": "SELECT",
+  "Original": "select count(*) a from user having a = '1'",
+  "Instructions": {
+    "OperatorType": "Filter",
+    "Predicate": "a = '1'",
+    "Inputs": [
+      {
+        "OperatorType": "Aggregate",
+        "Variant": "Ordered",
+        "Aggregates": "count(0) AS a",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "SelectScatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select count(*) as a from `user` where 1 != 1",
+            "Query": "select count(*) as a from `user`",
+            "Table": "`user`"
+          }
+        ]
+      }
+    ]
+  }
+}
+
+# Column and Literal not equal filter on scatter aggregates
+"select count(*) a from user having a != 10"
+"unsupported: filtering on results of aggregates"
+{
+  "QueryType": "SELECT",
+  "Original": "select count(*) a from user having a != 10",
+  "Instructions": {
+    "OperatorType": "Filter",
+    "Predicate": "a != 10",
+    "Inputs": [
+      {
+        "OperatorType": "Aggregate",
+        "Variant": "Ordered",
+        "Aggregates": "count(0) AS a",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "SelectScatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select count(*) as a from `user` where 1 != 1",
+            "Query": "select count(*) as a from `user`",
+            "Table": "`user`"
+          }
+        ]
+      }
+    ]
+  }
+}
+
+# Not equal filter with column and string literal on scatter aggregates
+"select count(*) a from user having a != '1'"
+"unsupported: filtering on results of aggregates"
+{
+  "QueryType": "SELECT",
+  "Original": "select count(*) a from user having a != '1'",
+  "Instructions": {
+    "OperatorType": "Filter",
+    "Predicate": "a != '1'",
+    "Inputs": [
+      {
+        "OperatorType": "Aggregate",
+        "Variant": "Ordered",
+        "Aggregates": "count(0) AS a",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "SelectScatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select count(*) as a from `user` where 1 != 1",
+            "Query": "select count(*) as a from `user`",
+            "Table": "`user`"
+          }
+        ]
+      }
+    ]
+  }
+}
+
+# Greater than filter on scatter aggregates
+"select count(*) a from user having a > 10"
+"unsupported: filtering on results of aggregates"
+{
+  "QueryType": "SELECT",
+  "Original": "select count(*) a from user having a \u003e 10",
+  "Instructions": {
+    "OperatorType": "Filter",
+    "Predicate": "a \u003e 10",
+    "Inputs": [
+      {
+        "OperatorType": "Aggregate",
+        "Variant": "Ordered",
+        "Aggregates": "count(0) AS a",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "SelectScatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select count(*) as a from `user` where 1 != 1",
+            "Query": "select count(*) as a from `user`",
+            "Table": "`user`"
+          }
+        ]
+      }
+    ]
+  }
+}
+
+# Greater Equal filter on scatter aggregates
+"select count(*) a from user having a >= 10"
+"unsupported: filtering on results of aggregates"
+{
+  "QueryType": "SELECT",
+  "Original": "select count(*) a from user having a \u003e= 10",
+  "Instructions": {
+    "OperatorType": "Filter",
+    "Predicate": "a \u003e= 10",
+    "Inputs": [
+      {
+        "OperatorType": "Aggregate",
+        "Variant": "Ordered",
+        "Aggregates": "count(0) AS a",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "SelectScatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select count(*) as a from `user` where 1 != 1",
+            "Query": "select count(*) as a from `user`",
+            "Table": "`user`"
+          }
+        ]
+      }
+    ]
+  }
+}
+
+# Less than filter on scatter aggregates
+"select count(*) a from user having a < 10"
+"unsupported: filtering on results of aggregates"
+{
+  "QueryType": "SELECT",
+  "Original": "select count(*) a from user having a \u003c 10",
+  "Instructions": {
+    "OperatorType": "Filter",
+    "Predicate": "a \u003c 10",
+    "Inputs": [
+      {
+        "OperatorType": "Aggregate",
+        "Variant": "Ordered",
+        "Aggregates": "count(0) AS a",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "SelectScatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select count(*) as a from `user` where 1 != 1",
+            "Query": "select count(*) as a from `user`",
+            "Table": "`user`"
+          }
+        ]
+      }
+    ]
+  }
+}
+
+# Less Equal filter on scatter aggregates
+"select count(*) a from user having a <= 10"
+"unsupported: filtering on results of aggregates"
+{
+  "QueryType": "SELECT",
+  "Original": "select count(*) a from user having a \u003c= 10",
+  "Instructions": {
+    "OperatorType": "Filter",
+    "Predicate": "a \u003c= 10",
+    "Inputs": [
+      {
+        "OperatorType": "Aggregate",
+        "Variant": "Ordered",
+        "Aggregates": "count(0) AS a",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "SelectScatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select count(*) as a from `user` where 1 != 1",
+            "Query": "select count(*) as a from `user`",
+            "Table": "`user`"
+          }
+        ]
+      }
+    ]
+  }
+}
+
+# Equal filter with hexadecimal value
+"select count(*) a from user having a = 0x01"
+"unsupported: filtering on results of aggregates"
+Gen4 error: expr cannot be converted, not supported: hexadecimal value: 0x01
diff --git a/go/vt/vtgate/planbuilder/testdata/select_cases.txt b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
index 372a17ddf4..55aa77d7ed 100644
--- a/go/vt/vtgate/planbuilder/testdata/select_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
@@ -2581,6 +2581,51 @@ Gen4 plan same as above
   }
 }
 
+# correlated subquery in exists clause with an order by
+"select col from user where exists(select user_id from user_extra where user_id = 3 and user_id < user.id) order by col"
+"unsupported: cross-shard correlated subquery"
+{
+  "QueryType": "SELECT",
+  "Original": "select col from user where exists(select user_id from user_extra where user_id = 3 and user_id \u003c user.id) order by col",
+  "Instructions": {
+    "OperatorType": "SemiJoin",
+    "JoinVars": {
+      "user_id": 0
+    },
+    "ProjectedIndexes": "-2",
+    "TableName": "`user`_user_extra",
+    "Inputs": [
+      {
+        "OperatorType": "Route",
+        "Variant": "SelectScatter",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select `user`.id, col, weight_string(col) from `user` where 1 != 1",
+        "OrderBy": "(1|2) ASC",
+        "Query": "select `user`.id, col, weight_string(col) from `user` order by col asc",
+        "Table": "`user`"
+      },
+      {
+        "OperatorType": "Route",
+        "Variant": "SelectEqualUnique",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select 1 from user_extra where 1 != 1",
+        "Query": "select 1 from user_extra where user_id = 3 and user_id \u003c :user_id",
+        "Table": "user_extra",
+        "Values": [
+          3
+        ],
+        "Vindex": "user_index"
+      }
+    ]
+  }
+}
+
 # correlated subquery having dependencies on two tables
 "select 1 from user u1, user u2 where exists (select 1 from user_extra ue where ue.col = u1.col and ue.col = u2.col)"
 "unsupported: cross-shard correlated subquery"
diff --git a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
index d787040745..d67579258e 100644
--- a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
@@ -83,11 +83,6 @@ Gen4 plan same as above
 "unsupported: '*' expression in cross-shard query"
 Gen4 error: cannot use column offsets in group statement when using `*`
 
-# Filtering on scatter aggregates
-"select count(*) a from user having a >10"
-"unsupported: filtering on results of aggregates"
-Gen4 plan same as above
-
 # group by must reference select list
 "select a from user group by b"
 "unsupported: in scatter query: group by column must reference column in SELECT list"
@@ -505,11 +500,11 @@ Gen4 plan same as above
 # TODO this should be planned without using OA and MS
 "select u.id from user u join user_extra ue on ue.id = u.id group by u.id having count(u.name) = 3"
 "unsupported: cross-shard query with aggregates"
-Gen4 error: unsupported: filtering on results of aggregates
+Gen4 error: expr cannot be converted, not supported: *sqlparser.FuncExpr
 
 "select (select 1 from user u having count(ue.col) > 10) from user_extra ue"
 "symbol ue.col not found in subquery"
-Gen4 error: unsupported: filtering on results of aggregates
+Gen4 error: expr cannot be converted, not supported: *sqlparser.FuncExpr
 
 # aggregation filtering by having on a route with no group by
 "select 1 from user having count(id) = 10"
@@ -528,7 +523,7 @@ Gen4 error: unsupported: filtering on results of aggregates
     "Table": "`user`"
   }
 }
-Gen4 error: unsupported: filtering on results of aggregates
+Gen4 error: expr cannot be converted, not supported: *sqlparser.FuncExpr
 
 # aggregation filtering by having on a route with no group by with non-unique vindex filter
 "select 1 from user having count(id) = 10 and name = 'a'"
@@ -551,7 +546,7 @@ Gen4 error: unsupported: filtering on results of aggregates
     "Vindex": "name_user_map"
   }
 }
-Gen4 error: unsupported: filtering on results of aggregates
+Gen4 error: expr cannot be converted, not supported: *sqlparser.FuncExpr
 
 # subquery of information_schema with itself and star expression in outer select
 "select a.*, u.id from information_schema.a a, user u where a.id in (select * from information_schema.b)"