Repository: vitessio/vitess
Pull request #10361 (2022-05-24 11:20:43)

Aggregation on top of LIMIT
## Description
Before this PR, the only way vtgate could do aggregation was to delegate most of the actual work down to mysql. There are a few situations where this is not really feasible to do. The solution was to make it possible to do all of the aggregation at the vtgate level.

We already had good support for all aggregation functions except `count`. That one was not done on the vtgate side, since we before assumed that we had been able to push down some of the `count`ing, and in that case we just need a `sum` on the vtgate side.

With this change, when vtgate is unable to push down aggregation, it can still perform the query, by fetching the necessary data and performing the grouping and aggregating on the vtgate.

## Related Issue(s)

## Checklist
-   [x] "Backport me!" label has been added if this change should be backported
-   [x] Tests were added or are not required
-   [x] Documentation was added or is not required

## Deployment Notes


github-actions[bot]: ### Review Checklist

Hello reviewers! :wave: Please follow this checklist when reviewing this Pull Request.

#### General
- [ ] Ensure that the Pull Request has the correct `release notes` label. `release notes none` should only be used for PRs that are so trivial that they need not be included.
- [ ] If a new flag is being introduced, review whether it is really needed. The flag names should be clear and intuitive (as far as possible), and the flag's help should be descriptive.

#### Bug fixes
- [ ] There should be at least one unit or end-to-end test.
- [ ] The Pull Request description should either include a link to an issue that describes the bug OR an actual description of the bug and how to reproduce, along with a description of the fix.

#### Non-trivial changes
- [ ] There should be some code comments as to why things are implemented the way they are.

#### New/Existing features
- [ ] Should be documented, either by modifying the existing documentation or creating new documentation.
- [ ] New features should have a link to a feature request issue or an RFC that documents the use cases, corner cases and test cases.

#### Backward compatibility            
- [ ] Protobuf changes should be wire-compatible.
- [ ] Changes to `_vt` tables and RPCs need to be backward compatible.
- [ ] `vtctl` command output order should be stable and `awk`-able.

Patch:
diff --git a/go/vt/vtgate/engine/ordered_aggregate.go b/go/vt/vtgate/engine/ordered_aggregate.go
index dc159a4970..7878004ae0 100644
--- a/go/vt/vtgate/engine/ordered_aggregate.go
+++ b/go/vt/vtgate/engine/ordered_aggregate.go
@@ -43,10 +43,13 @@ var _ Primitive = (*OrderedAggregate)(nil)
 type OrderedAggregate struct {
 	// PreProcess is true if one of the aggregates needs preprocessing.
 	PreProcess bool `json:",omitempty"`
+
 	// Aggregates specifies the aggregation parameters for each
 	// aggregation function: function opcode and input column number.
 	Aggregates []*AggregateParams
 
+	AggrOnEngine bool
+
 	// GroupByKeys specifies the input values that must be used for
 	// the aggregation key.
 	GroupByKeys []*GroupByParams
@@ -112,7 +115,7 @@ func (ap *AggregateParams) isDistinct() bool {
 }
 
 func (ap *AggregateParams) preProcess() bool {
-	return ap.Opcode == AggregateCountDistinct || ap.Opcode == AggregateSumDistinct || ap.Opcode == AggregateGtid
+	return ap.Opcode == AggregateCountDistinct || ap.Opcode == AggregateSumDistinct || ap.Opcode == AggregateGtid || ap.Opcode == AggregateCount
 }
 
 func (ap *AggregateParams) String() string {
@@ -143,6 +146,7 @@ const (
 	AggregateSumDistinct
 	AggregateGtid
 	AggregateRandom
+	AggregateCountStar
 )
 
 var (
@@ -150,6 +154,7 @@ var (
 	OpcodeType = map[AggregateOpcode]querypb.Type{
 		AggregateCountDistinct: sqltypes.Int64,
 		AggregateCount:         sqltypes.Int64,
+		AggregateCountStar:     sqltypes.Int64,
 		AggregateSumDistinct:   sqltypes.Decimal,
 		AggregateSum:           sqltypes.Decimal,
 		AggregateGtid:          sqltypes.VarChar,
@@ -172,6 +177,8 @@ var SupportedAggregates = map[string]AggregateOpcode{
 	"count_distinct": AggregateCountDistinct,
 	"sum_distinct":   AggregateSumDistinct,
 	"vgtid":          AggregateGtid,
+	"count_star":     AggregateCountStar,
+	"random":         AggregateRandom,
 }
 
 func (code AggregateOpcode) String() string {
@@ -180,7 +187,7 @@ func (code AggregateOpcode) String() string {
 			return k
 		}
 	}
-	return "random"
+	return "ERROR"
 }
 
 // MarshalJSON serializes the AggregateOpcode as a JSON string.
@@ -224,7 +231,7 @@ func (oa *OrderedAggregate) execute(vcursor VCursor, bindVars map[string]*queryp
 		return nil, err
 	}
 	out := &sqltypes.Result{
-		Fields: convertFields(result.Fields, oa.PreProcess, oa.Aggregates),
+		Fields: convertFields(result.Fields, oa.PreProcess, oa.Aggregates, oa.AggrOnEngine),
 		Rows:   make([][]sqltypes.Value, 0, len(result.Rows)),
 	}
 	// This code is similar to the one in StreamExecute.
@@ -232,7 +239,7 @@ func (oa *OrderedAggregate) execute(vcursor VCursor, bindVars map[string]*queryp
 	var curDistincts []sqltypes.Value
 	for _, row := range result.Rows {
 		if current == nil {
-			current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates)
+			current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates, oa.AggrOnEngine)
 			continue
 		}
 		equal, err := oa.keysEqual(current, row, oa.Collations)
@@ -248,7 +255,7 @@ func (oa *OrderedAggregate) execute(vcursor VCursor, bindVars map[string]*queryp
 			continue
 		}
 		out.Rows = append(out.Rows, current)
-		current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates)
+		current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates, oa.AggrOnEngine)
 	}
 
 	if current != nil {
@@ -273,7 +280,7 @@ func (oa *OrderedAggregate) TryStreamExecute(vcursor VCursor, bindVars map[strin
 
 	err := vcursor.StreamExecutePrimitive(oa.Input, bindVars, wantfields, func(qr *sqltypes.Result) error {
 		if len(qr.Fields) != 0 {
-			fields = convertFields(qr.Fields, oa.PreProcess, oa.Aggregates)
+			fields = convertFields(qr.Fields, oa.PreProcess, oa.Aggregates, oa.AggrOnEngine)
 			if err := cb(&sqltypes.Result{Fields: fields}); err != nil {
 				return err
 			}
@@ -281,7 +288,7 @@ func (oa *OrderedAggregate) TryStreamExecute(vcursor VCursor, bindVars map[strin
 		// This code is similar to the one in Execute.
 		for _, row := range qr.Rows {
 			if current == nil {
-				current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates)
+				current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates, oa.AggrOnEngine)
 				continue
 			}
 
@@ -300,7 +307,7 @@ func (oa *OrderedAggregate) TryStreamExecute(vcursor VCursor, bindVars map[strin
 			if err := cb(&sqltypes.Result{Rows: [][]sqltypes.Value{current}}); err != nil {
 				return err
 			}
-			current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates)
+			current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates, oa.AggrOnEngine)
 		}
 		return nil
 	})
@@ -316,12 +323,12 @@ func (oa *OrderedAggregate) TryStreamExecute(vcursor VCursor, bindVars map[strin
 	return nil
 }
 
-func convertFields(fields []*querypb.Field, preProcess bool, aggrs []*AggregateParams) []*querypb.Field {
+func convertFields(fields []*querypb.Field, preProcess bool, aggrs []*AggregateParams, aggrOnEngine bool) []*querypb.Field {
 	if !preProcess {
 		return fields
 	}
 	for _, aggr := range aggrs {
-		if !aggr.preProcess() {
+		if !aggr.preProcess() && !aggrOnEngine {
 			continue
 		}
 		fields[aggr.Col] = &querypb.Field{
@@ -335,7 +342,7 @@ func convertFields(fields []*querypb.Field, preProcess bool, aggrs []*AggregateP
 	return fields
 }
 
-func convertRow(row []sqltypes.Value, preProcess bool, aggregates []*AggregateParams) (newRow []sqltypes.Value, curDistincts []sqltypes.Value) {
+func convertRow(row []sqltypes.Value, preProcess bool, aggregates []*AggregateParams, aggrOnEngine bool) (newRow []sqltypes.Value, curDistincts []sqltypes.Value) {
 	if !preProcess {
 		return row, nil
 	}
@@ -343,6 +350,14 @@ func convertRow(row []sqltypes.Value, preProcess bool, aggregates []*AggregatePa
 	curDistincts = make([]sqltypes.Value, len(aggregates))
 	for index, aggr := range aggregates {
 		switch aggr.Opcode {
+		case AggregateCountStar:
+			newRow[aggr.Col] = countOne
+		case AggregateCount:
+			val := countOne
+			if row[aggr.Col].IsNull() {
+				val = countZero
+			}
+			newRow[aggr.Col] = val
 		case AggregateCountDistinct:
 			curDistincts[index] = findComparableCurrentDistinct(row, aggr)
 			// Type is int64. Ok to call MakeTrusted.
@@ -351,6 +366,18 @@ func convertRow(row []sqltypes.Value, preProcess bool, aggregates []*AggregatePa
 			} else {
 				newRow[aggr.Col] = countOne
 			}
+		case AggregateSum:
+			if !aggrOnEngine {
+				break
+			}
+			if row[aggr.Col].IsNull() {
+				break
+			}
+			var err error
+			newRow[aggr.Col], err = evalengine.Cast(row[aggr.Col], OpcodeType[aggr.Opcode])
+			if err != nil {
+				newRow[aggr.Col] = sumZero
+			}
 		case AggregateSumDistinct:
 			curDistincts[index] = findComparableCurrentDistinct(row, aggr)
 			var err error
@@ -388,7 +415,7 @@ func (oa *OrderedAggregate) GetFields(vcursor VCursor, bindVars map[string]*quer
 	if err != nil {
 		return nil, err
 	}
-	qr = &sqltypes.Result{Fields: convertFields(qr.Fields, oa.PreProcess, oa.Aggregates)}
+	qr = &sqltypes.Result{Fields: convertFields(qr.Fields, oa.PreProcess, oa.Aggregates, oa.AggrOnEngine)}
 	return qr.Truncate(oa.TruncateColumnCount), nil
 }
 
@@ -448,9 +475,22 @@ func merge(
 		}
 		var err error
 		switch aggr.Opcode {
-		case AggregateCount, AggregateSum:
+		case AggregateCountStar:
+			value := row1[aggr.Col]
+			result[aggr.Col], err = evalengine.NullSafeAdd(value, countOne, fields[aggr.Col].Type)
+		case AggregateCount:
+			val := countOne
+			if row2[aggr.Col].IsNull() {
+				val = countZero
+			}
+			result[aggr.Col], err = evalengine.NullSafeAdd(row1[aggr.Col], val, fields[aggr.Col].Type)
+		case AggregateSum:
 			value := row1[aggr.Col]
 			v2 := row2[aggr.Col]
+			if value.IsNull() && v2.IsNull() {
+				result[aggr.Col] = sqltypes.NULL
+				break
+			}
 			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])
diff --git a/go/vt/vtgate/engine/projection.go b/go/vt/vtgate/engine/projection.go
index 7ffb4a9b77..97a080be2f 100644
--- a/go/vt/vtgate/engine/projection.go
+++ b/go/vt/vtgate/engine/projection.go
@@ -80,33 +80,11 @@ func (p *Projection) TryExecute(vcursor VCursor, bindVars map[string]*querypb.Bi
 }
 
 // TryStreamExecute implements the Primitive interface
-func (p *Projection) TryStreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantields bool, callback func(*sqltypes.Result) error) error {
-	result, err := vcursor.ExecutePrimitive(p.Input, bindVars, wantields)
+func (p *Projection) TryStreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
+	result, err := p.TryExecute(vcursor, bindVars, wantfields)
 	if err != nil {
 		return err
 	}
-
-	env := evalengine.EnvWithBindVars(bindVars, vcursor.ConnCollation())
-	if wantields {
-		err = p.addFields(env, result)
-		if err != nil {
-			return err
-		}
-	}
-	var rows [][]sqltypes.Value
-	env.Fields = result.Fields
-	for _, row := range result.Rows {
-		env.Row = row
-		for _, exp := range p.Exprs {
-			result, err := env.Evaluate(exp)
-			if err != nil {
-				return err
-			}
-			row = append(row, result.Value())
-		}
-		rows = append(rows, row)
-	}
-	result.Rows = rows
 	return callback(result)
 }
 
diff --git a/go/vt/vtgate/engine/scalar_aggregation.go b/go/vt/vtgate/engine/scalar_aggregation.go
index c6bd02d2af..642cee13f8 100644
--- a/go/vt/vtgate/engine/scalar_aggregation.go
+++ b/go/vt/vtgate/engine/scalar_aggregation.go
@@ -17,6 +17,8 @@ limitations under the License.
 package engine
 
 import (
+	"sync"
+
 	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/sqltypes"
 	querypb "vitess.io/vitess/go/vt/proto/query"
@@ -30,6 +32,9 @@ var _ Primitive = (*ScalarAggregate)(nil)
 type ScalarAggregate struct {
 	// PreProcess is true if one of the aggregates needs preprocessing.
 	PreProcess bool `json:",omitempty"`
+
+	AggrOnEngine bool
+
 	// Aggregates specifies the aggregation parameters for each
 	// aggregation function: function opcode and input column number.
 	Aggregates []*AggregateParams
@@ -69,7 +74,7 @@ func (sa *ScalarAggregate) GetFields(vcursor VCursor, bindVars map[string]*query
 	if err != nil {
 		return nil, err
 	}
-	qr = &sqltypes.Result{Fields: convertFields(qr.Fields, sa.PreProcess, sa.Aggregates)}
+	qr = &sqltypes.Result{Fields: convertFields(qr.Fields, sa.PreProcess, sa.Aggregates, sa.AggrOnEngine)}
 	return qr.Truncate(sa.TruncateColumnCount), nil
 }
 
@@ -85,14 +90,14 @@ func (sa *ScalarAggregate) TryExecute(vcursor VCursor, bindVars map[string]*quer
 		return nil, err
 	}
 	out := &sqltypes.Result{
-		Fields: convertFields(result.Fields, sa.PreProcess, sa.Aggregates),
+		Fields: convertFields(result.Fields, sa.PreProcess, sa.Aggregates, sa.AggrOnEngine),
 	}
 
 	var resultRow []sqltypes.Value
 	var curDistincts []sqltypes.Value
 	for _, row := range result.Rows {
 		if resultRow == nil {
-			resultRow, curDistincts = convertRow(row, sa.PreProcess, sa.Aggregates)
+			resultRow, curDistincts = convertRow(row, sa.PreProcess, sa.Aggregates, sa.AggrOnEngine)
 			continue
 		}
 		resultRow, curDistincts, err = merge(result.Fields, resultRow, row, curDistincts, sa.Collations, sa.Aggregates)
@@ -128,10 +133,16 @@ func (sa *ScalarAggregate) TryStreamExecute(vcursor VCursor, bindVars map[string
 	var curDistincts []sqltypes.Value
 	var fields []*querypb.Field
 	fieldsSent := false
+	var mu sync.Mutex
 
 	err := vcursor.StreamExecutePrimitive(sa.Input, bindVars, wantfields, func(result *sqltypes.Result) error {
+		// as the underlying primitive call is not sync
+		// and here scalar aggregate is using shared variables we have to sync the callback
+		// for correct aggregation.
+		mu.Lock()
+		defer mu.Unlock()
 		if len(result.Fields) != 0 && !fieldsSent {
-			fields = convertFields(result.Fields, sa.PreProcess, sa.Aggregates)
+			fields = convertFields(result.Fields, sa.PreProcess, sa.Aggregates, sa.AggrOnEngine)
 			if err := cb(&sqltypes.Result{Fields: fields}); err != nil {
 				return err
 			}
@@ -141,7 +152,7 @@ func (sa *ScalarAggregate) TryStreamExecute(vcursor VCursor, bindVars map[string
 		// this code is very similar to the TryExecute method
 		for _, row := range result.Rows {
 			if current == nil {
-				current, curDistincts = convertRow(row, sa.PreProcess, sa.Aggregates)
+				current, curDistincts = convertRow(row, sa.PreProcess, sa.Aggregates, sa.AggrOnEngine)
 				continue
 			}
 			var err error
@@ -176,7 +187,8 @@ func createEmptyValueFor(opcode AggregateOpcode) (sqltypes.Value, error) {
 	switch opcode {
 	case
 		AggregateCountDistinct,
-		AggregateCount:
+		AggregateCount,
+		AggregateCountStar:
 		return countZero, nil
 	case
 		AggregateSumDistinct,
diff --git a/go/vt/vtgate/planbuilder/abstract/queryprojection.go b/go/vt/vtgate/planbuilder/abstract/queryprojection.go
index 7ba9fde31a..068603dafe 100644
--- a/go/vt/vtgate/planbuilder/abstract/queryprojection.go
+++ b/go/vt/vtgate/planbuilder/abstract/queryprojection.go
@@ -452,6 +452,28 @@ func (qp *QueryProjection) NeedsDistinct() bool {
 }
 
 func (qp *QueryProjection) AggregationExpressions() (out []Aggr, err error) {
+orderBy:
+	for _, orderExpr := range qp.OrderExprs {
+		if qp.isOrderByExprInGroupBy(orderExpr) {
+			continue orderBy
+		}
+		orderExpr := orderExpr.Inner.Expr
+		for _, expr := range qp.SelectExprs {
+			col, ok := expr.Col.(*sqlparser.AliasedExpr)
+			if !ok {
+				continue
+			}
+			if sqlparser.EqualsExpr(col.Expr, orderExpr) {
+				continue orderBy // we found the expression we were looking for!
+			}
+		}
+		qp.SelectExprs = append(qp.SelectExprs, SelectExpr{
+			Col:  &sqlparser.AliasedExpr{Expr: orderExpr},
+			Aggr: sqlparser.ContainsAggregation(orderExpr),
+		})
+		qp.AddedColumn++
+	}
+
 	for idx, expr := range qp.SelectExprs {
 		aliasedExpr, err := expr.GetAliasedExpr()
 		if err != nil {
@@ -483,6 +505,10 @@ func (qp *QueryProjection) AggregationExpressions() (out []Aggr, err error) {
 			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: in scatter query: aggregation function '%s'", funcName)
 		}
 
+		if _, isStar := fExpr.Exprs[0].(*sqlparser.StarExpr); isStar && opcode == engine.AggregateCount {
+			opcode = engine.AggregateCountStar
+		}
+
 		if fExpr.Distinct {
 			switch opcode {
 			case engine.AggregateCount:
@@ -500,28 +526,6 @@ func (qp *QueryProjection) AggregationExpressions() (out []Aggr, err error) {
 			Distinct: fExpr.Distinct,
 		})
 	}
-	for i, orderExpr := range qp.OrderExprs {
-		newIdx := i + len(qp.SelectExprs)
-		if !qp.isOrderByExprInGroupBy(orderExpr) {
-
-			original := &sqlparser.AliasedExpr{Expr: orderExpr.Inner.Expr}
-			found := false
-			for _, aggr := range out {
-				if sqlparser.EqualsRefOfAliasedExpr(aggr.Original, original) {
-					found = true
-					break
-				}
-			}
-			if found {
-				continue
-			}
-			out = append(out, Aggr{
-				Original: original,
-				OpCode:   engine.AggregateRandom,
-				Index:    &newIdx,
-			})
-		}
-	}
 	return
 }
 
diff --git a/go/vt/vtgate/planbuilder/aggregation_pushing.go b/go/vt/vtgate/planbuilder/aggregation_pushing.go
index cb937910d8..7f496ab1f0 100644
--- a/go/vt/vtgate/planbuilder/aggregation_pushing.go
+++ b/go/vt/vtgate/planbuilder/aggregation_pushing.go
@@ -26,13 +26,21 @@ import (
 )
 
 // pushAggregation pushes grouping and aggregation as far down in the tree as possible
+// the output `outputAggrsOffset` needs a little explaining: this is the offsets for aggregation - remember
+// that aggregation can be broken down into multiple expressions that are later combined.
+// this is why this output is a slice of slices
 func (hp *horizonPlanning) pushAggregation(
 	ctx *plancontext.PlanningContext,
 	plan logicalPlan,
 	grouping []abstract.GroupBy,
 	aggregations []abstract.Aggr,
 	ignoreOutputOrder bool,
-) (output logicalPlan, groupingOffsets []offsets, outputAggrsOffset [][]offsets, err error) {
+) (output logicalPlan,
+	groupingOffsets []offsets,
+	outputAggrsOffset [][]offsets,
+	pushed bool,
+	err error) {
+	pushed = true
 	switch plan := plan.(type) {
 	case *routeGen4:
 		output = plan
@@ -46,13 +54,52 @@ func (hp *horizonPlanning) pushAggregation(
 
 	case *semiJoin:
 		output = plan
-		groupingOffsets, outputAggrsOffset, err = hp.pushAggrOnSemiJoin(ctx, plan, grouping, aggregations, ignoreOutputOrder)
+		groupingOffsets, outputAggrsOffset, pushed, err = hp.pushAggrOnSemiJoin(ctx, plan, grouping, aggregations, ignoreOutputOrder)
 		return
 
 	case *simpleProjection:
 		// we just remove the simpleProjection. We are doing an OA on top anyway, so no need to clean up the output columns
 		return hp.pushAggregation(ctx, plan.input, grouping, aggregations, ignoreOutputOrder)
 
+	case *limit:
+		// if we are seeing a limit, it's because we are building on top of a derived table.
+		output = plan
+		pushed = false
+
+		for _, grp := range grouping {
+			offset, wOffset, err := wrapAndPushExpr(ctx, grp.Inner, grp.WeightStrExpr, plan.input)
+			if err != nil {
+				return nil, nil, nil, false, err
+			}
+			groupingOffsets = append(groupingOffsets, offsets{
+				col:   offset,
+				wsCol: wOffset,
+			})
+		}
+
+		for _, aggr := range aggregations {
+			var offset int
+			fExpr, ok := aggr.Original.Expr.(*sqlparser.FuncExpr)
+			if !ok {
+				return nil, nil, nil, false, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "[BUG]: unexpected expression: %v", aggr.Original)
+			}
+			if len(fExpr.Exprs) != 1 {
+				return nil, nil, nil, false, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "[BUG]: unexpected expression: %v", fExpr)
+			}
+			switch e := fExpr.Exprs[0].(type) {
+			case *sqlparser.StarExpr:
+				offset = 0
+			case *sqlparser.AliasedExpr:
+				offset, _, err = pushProjection(ctx, e, plan.input, true, true, false)
+			}
+			if err != nil {
+				return nil, nil, nil, false, err
+			}
+
+			outputAggrsOffset = append(outputAggrsOffset, []offsets{newOffset(offset)})
+		}
+
+		return
 	default:
 		err = vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "using aggregation on top of a %T plan is not yet supported", plan)
 		return
@@ -184,7 +231,7 @@ func countStarAggr() *abstract.Aggr {
 
 	return &abstract.Aggr{
 		Original: &sqlparser.AliasedExpr{Expr: f},
-		OpCode:   engine.AggregateCount,
+		OpCode:   engine.AggregateCountStar,
 		Alias:    "count(*)",
 	}
 }
@@ -228,12 +275,12 @@ func (hp *horizonPlanning) pushAggrOnJoin(
 	}
 
 	// Next we push the aggregations to both sides
-	newLHS, lhsOffsets, lhsAggrOffsets, err := hp.filteredPushAggregation(ctx, join.Left, lhsGrouping, lhsAggrs, true)
+	newLHS, lhsOffsets, lhsAggrOffsets, _, err := hp.filteredPushAggregation(ctx, join.Left, lhsGrouping, lhsAggrs, true)
 	if err != nil {
 		return nil, nil, err
 	}
 
-	newRHS, rhsOffsets, rhsAggrOffsets, err := hp.filteredPushAggregation(ctx, join.Right, rhsGrouping, rhsAggrs, true)
+	newRHS, rhsOffsets, rhsAggrOffsets, _, err := hp.filteredPushAggregation(ctx, join.Right, rhsGrouping, rhsAggrs, true)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -296,18 +343,18 @@ func (hp *horizonPlanning) pushAggrOnSemiJoin(
 	grouping []abstract.GroupBy,
 	aggregations []abstract.Aggr,
 	ignoreOutputOrder bool,
-) ([]offsets, [][]offsets, error) {
+) ([]offsets, [][]offsets, bool, error) {
 	// We need to group by the columns used in the join condition.
 	// If we don't, the LHS will not be able to return the column, and it can't be used to send down to the RHS
 	lhsCols, err := hp.createGroupingsForColumns(join.LHSColumns)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, false, err
 	}
 
 	totalGrouping := append(grouping, lhsCols...)
-	newLeft, groupingOffsets, aggrParams, err := hp.pushAggregation(ctx, join.lhs, totalGrouping, aggregations, ignoreOutputOrder)
+	newLeft, groupingOffsets, aggrParams, pushed, err := hp.pushAggregation(ctx, join.lhs, totalGrouping, aggregations, ignoreOutputOrder)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, false, err
 	}
 	join.lhs = newLeft
 
@@ -316,7 +363,7 @@ func (hp *horizonPlanning) pushAggrOnSemiJoin(
 		outputGroupings = append(outputGroupings, groupingOffsets[idx])
 	}
 
-	return outputGroupings, aggrParams, nil
+	return outputGroupings, aggrParams, pushed, nil
 }
 
 // this method takes a slice of aggregations that can have missing spots in the form of `nil`,
@@ -331,7 +378,7 @@ func (hp *horizonPlanning) filteredPushAggregation(
 	grouping []abstract.GroupBy,
 	aggregations []*abstract.Aggr,
 	ignoreOutputOrder bool,
-) (out logicalPlan, groupingOffsets []offsets, outputAggrs [][]offsets, err error) {
+) (out logicalPlan, groupingOffsets []offsets, outputAggrs [][]offsets, pushed bool, err error) {
 	used := make([]bool, len(aggregations))
 	var aggrs []abstract.Aggr
 
@@ -341,9 +388,9 @@ func (hp *horizonPlanning) filteredPushAggregation(
 			aggrs = append(aggrs, *aggr)
 		}
 	}
-	newplan, groupingOffsets, pushedAggrs, err := hp.pushAggregation(ctx, plan, grouping, aggrs, ignoreOutputOrder)
+	newplan, groupingOffsets, pushedAggrs, pushed, err := hp.pushAggregation(ctx, plan, grouping, aggrs, ignoreOutputOrder)
 	if err != nil {
-		return nil, nil, nil, err
+		return nil, nil, nil, pushed, err
 	}
 	idx := 0
 	for _, b := range used {
@@ -354,7 +401,7 @@ func (hp *horizonPlanning) filteredPushAggregation(
 		outputAggrs = append(outputAggrs, pushedAggrs[idx])
 		idx++
 	}
-	return newplan, groupingOffsets, outputAggrs, nil
+	return newplan, groupingOffsets, outputAggrs, pushed, nil
 }
 
 func isMinOrMax(in engine.AggregateOpcode) bool {
diff --git a/go/vt/vtgate/planbuilder/horizon_planning.go b/go/vt/vtgate/planbuilder/horizon_planning.go
index 81945b1f41..3bd8679b13 100644
--- a/go/vt/vtgate/planbuilder/horizon_planning.go
+++ b/go/vt/vtgate/planbuilder/horizon_planning.go
@@ -20,7 +20,6 @@ import (
 	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/abstract"
-	"vitess.io/vitess/go/vt/vtgate/planbuilder/physical"
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
 
 	"vitess.io/vitess/go/vt/vtgate/semantics"
@@ -172,266 +171,6 @@ func (hp *horizonPlanning) truncateColumnsIfNeeded(ctx *plancontext.PlanningCont
 	return plan, nil
 }
 
-// pushProjection pushes a projection to the plan.
-func pushProjection(
-	ctx *plancontext.PlanningContext,
-	expr *sqlparser.AliasedExpr,
-	plan logicalPlan,
-	inner, reuseCol, hasAggregation bool,
-) (offset int, added bool, err error) {
-	switch node := plan.(type) {
-	case *routeGen4:
-		return addExpressionToRoute(ctx, node, expr, reuseCol)
-	case *hashJoin:
-		lhsSolves := node.Left.ContainsTables()
-		rhsSolves := node.Right.ContainsTables()
-		deps := ctx.SemTable.RecursiveDeps(expr.Expr)
-		var column int
-		var appended bool
-		passDownReuseCol := reuseCol
-		if !reuseCol {
-			passDownReuseCol = expr.As.IsEmpty()
-		}
-		switch {
-		case deps.IsSolvedBy(lhsSolves):
-			offset, added, err := pushProjection(ctx, expr, node.Left, inner, passDownReuseCol, hasAggregation)
-			if err != nil {
-				return 0, false, err
-			}
-			column = -(offset + 1)
-			appended = added
-		case deps.IsSolvedBy(rhsSolves):
-			offset, added, err := pushProjection(ctx, expr, node.Right, inner && node.Opcode != engine.LeftJoin, passDownReuseCol, hasAggregation)
-			if err != nil {
-				return 0, false, err
-			}
-			column = offset + 1
-			appended = added
-		default:
-			// if an expression has aggregation, then it should not be split up and pushed to both sides,
-			// for example an expression like count(*) will have dependencies on both sides, but we should not push it
-			// instead we should return an error
-			if hasAggregation {
-				return 0, false, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: cross-shard query with aggregates")
-			}
-			return 0, false, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: hash join with projection from both sides of the join")
-		}
-		if reuseCol && !appended {
-			for idx, col := range node.Cols {
-				if column == col {
-					return idx, false, nil
-				}
-			}
-			// the column was not appended to either child, but we could not find it in out cols list,
-			// so we'll still add it
-		}
-		node.Cols = append(node.Cols, column)
-		return len(node.Cols) - 1, true, nil
-	case *joinGen4:
-		lhsSolves := node.Left.ContainsTables()
-		rhsSolves := node.Right.ContainsTables()
-		deps := ctx.SemTable.RecursiveDeps(expr.Expr)
-		var column int
-		var appended bool
-		passDownReuseCol := reuseCol
-		if !reuseCol {
-			passDownReuseCol = expr.As.IsEmpty()
-		}
-		switch {
-		case deps.IsSolvedBy(lhsSolves):
-			offset, added, err := pushProjection(ctx, expr, node.Left, inner, passDownReuseCol, hasAggregation)
-			if err != nil {
-				return 0, false, err
-			}
-			column = -(offset + 1)
-			appended = added
-		case deps.IsSolvedBy(rhsSolves):
-			offset, added, err := pushProjection(ctx, expr, node.Right, inner && node.Opcode != engine.LeftJoin, passDownReuseCol, hasAggregation)
-			if err != nil {
-				return 0, false, err
-			}
-			column = offset + 1
-			appended = added
-		default:
-			// if an expression has aggregation, then it should not be split up and pushed to both sides,
-			// for example an expression like count(*) will have dependencies on both sides, but we should not push it
-			// instead we should return an error
-			if hasAggregation {
-				return 0, false, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: cross-shard query with aggregates")
-			}
-			// now we break the expression into left and right side dependencies and rewrite the left ones to bind variables
-			bvName, cols, rewrittenExpr, err := physical.BreakExpressionInLHSandRHS(ctx, expr.Expr, lhsSolves)
-			if err != nil {
-				return 0, false, err
-			}
-			// go over all the columns coming from the left side of the tree and push them down. While at it, also update the bind variable map.
-			// It is okay to reuse the columns on the left side since
-			// the final expression which will be selected will be pushed into the right side.
-			for i, col := range cols {
-				colOffset, _, err := pushProjection(ctx, &sqlparser.AliasedExpr{Expr: col}, node.Left, inner, true, false)
-				if err != nil {
-					return 0, false, err
-				}
-				node.Vars[bvName[i]] = colOffset
-			}
-			// push the rewritten expression on the right side of the tree. Here we should take care whether we want to reuse the expression or not.
-			expr.Expr = rewrittenExpr
-			offset, added, err := pushProjection(ctx, expr, node.Right, inner && node.Opcode != engine.LeftJoin, passDownReuseCol, false)
-			if err != nil {
-				return 0, false, err
-			}
-			column = offset + 1
-			appended = added
-		}
-		if reuseCol && !appended {
-			for idx, col := range node.Cols {
-				if column == col {
-					return idx, false, nil
-				}
-			}
-			// the column was not appended to either child, but we could not find it in out cols list,
-			// so we'll still add it
-		}
-		node.Cols = append(node.Cols, column)
-		return len(node.Cols) - 1, true, nil
-	case *simpleProjection:
-		offset, _, err := pushProjection(ctx, expr, node.input, inner, true, hasAggregation)
-		if err != nil {
-			return 0, false, err
-		}
-		for i, value := range node.eSimpleProj.Cols {
-			// we return early if we already have the column in the simple projection's
-			// output list so we do not add it again.
-			if reuseCol && value == offset {
-				return i, false, nil
-			}
-		}
-		node.eSimpleProj.Cols = append(node.eSimpleProj.Cols, offset)
-		return len(node.eSimpleProj.Cols) - 1, true, nil
-	case *orderedAggregate:
-		colName, isColName := expr.Expr.(*sqlparser.ColName)
-		for _, aggregate := range node.aggregates {
-			if sqlparser.EqualsExpr(aggregate.Expr, expr.Expr) {
-				return aggregate.Col, false, nil
-			}
-			if isColName && colName.Name.EqualString(aggregate.Alias) {
-				return aggregate.Col, false, nil
-			}
-		}
-		for _, key := range node.groupByKeys {
-			if sqlparser.EqualsExpr(key.Expr, expr.Expr) {
-				return key.KeyCol, false, nil
-			}
-		}
-		offset, _, err := pushProjection(ctx, expr, node.input, inner, true, hasAggregation)
-		if err != nil {
-			return 0, false, err
-		}
-		node.aggregates = append(node.aggregates, &engine.AggregateParams{
-			Opcode:   engine.AggregateRandom,
-			Col:      offset,
-			Alias:    expr.ColumnName(),
-			Expr:     expr.Expr,
-			Original: expr,
-		})
-		return offset, true, nil
-	case *vindexFunc:
-		colsBefore := len(node.eVindexFunc.Cols)
-		i, err := node.SupplyProjection(expr, reuseCol)
-		if err != nil {
-			return 0, false, err
-		}
-		return i /* col added */, len(node.eVindexFunc.Cols) > colsBefore, nil
-	case *limit, *projection, *pulloutSubquery, *distinct, *filter:
-		// All of these either push to the single source, or push to the LHS
-		src := node.Inputs()[0]
-		return pushProjection(ctx, expr, src, inner, reuseCol, hasAggregation)
-	case *semiJoin:
-		passDownReuseCol := reuseCol
-		if !reuseCol {
-			passDownReuseCol = expr.As.IsEmpty()
-		}
-		offset, added, err := pushProjection(ctx, expr, node.lhs, inner, passDownReuseCol, hasAggregation)
-		if err != nil {
-			return 0, false, err
-		}
-		column := -(offset + 1)
-		if reuseCol && !added {
-			for idx, col := range node.cols {
-				if column == col {
-					return idx, false, nil
-				}
-			}
-		}
-		node.cols = append(node.cols, column)
-		return len(node.cols) - 1, true, nil
-	case *concatenateGen4:
-		if hasAggregation {
-			return 0, false, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: aggregation on unions")
-		}
-		offset, added, err := pushProjection(ctx, expr, node.sources[0], inner, reuseCol, hasAggregation)
-		if err != nil {
-			return 0, false, err
-		}
-		if added && ctx.SemTable.DirectDeps(expr.Expr).NumberOfTables() > 0 {
-			return 0, false, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "pushing projection %v on concatenate should reference an existing column", sqlparser.String(expr))
-		}
-		if added {
-			for _, source := range node.sources[1:] {
-				_, _, err := pushProjection(ctx, expr, source, inner, reuseCol, hasAggregation)
-				if err != nil {
-					return 0, false, err
-				}
-			}
-		}
-		return offset, added, nil
-	default:
-		return 0, false, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "[BUG] push projection does not yet support: %T", node)
-	}
-}
-
-func addExpressionToRoute(ctx *plancontext.PlanningContext, rb *routeGen4, expr *sqlparser.AliasedExpr, reuseCol bool) (int, bool, error) {
-	if reuseCol {
-		if i := checkIfAlreadyExists(expr, rb.Select, ctx.SemTable); i != -1 {
-			return i, false, nil
-		}
-	}
-	expr.Expr = sqlparser.RemoveKeyspaceFromColName(expr.Expr)
-	sel, isSel := rb.Select.(*sqlparser.Select)
-	if !isSel {
-		return 0, false, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.BadFieldError, "unsupported: pushing projection '%s' on %T", sqlparser.String(expr), rb.Select)
-	}
-
-	if ctx.RewriteDerivedExpr {
-		// if we are trying to push a projection that belongs to a DerivedTable
-		// we rewrite that expression, so it matches the column name used inside
-		// that derived table.
-		err := rewriteProjectionOfDerivedTable(expr, ctx.SemTable)
-		if err != nil {
-			return 0, false, err
-		}
-	}
-
-	offset := len(sel.SelectExprs)
-	sel.SelectExprs = append(sel.SelectExprs, expr)
-	return offset, true, nil
-}
-
-func rewriteProjectionOfDerivedTable(expr *sqlparser.AliasedExpr, semTable *semantics.SemTable) error {
-	ti, err := semTable.TableInfoForExpr(expr.Expr)
-	if err != nil && err != semantics.ErrMultipleTables {
-		return err
-	}
-	_, isDerivedTable := ti.(*semantics.DerivedTable)
-	if isDerivedTable {
-		expr.Expr, err = semantics.RewriteDerivedExpression(expr.Expr, ti)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 func checkIfAlreadyExists(expr *sqlparser.AliasedExpr, node sqlparser.SelectStatement, semTable *semantics.SemTable) int {
 	exprDep := semTable.RecursiveDeps(expr.Expr)
 	// Here to find if the expr already exists in the SelectStatement, we have 3 cases
@@ -568,10 +307,14 @@ func (hp *horizonPlanning) planAggrUsingOA(
 		oa.preProcess = true
 	}
 
-	newPlan, groupingOffsets, aggrParamOffsets, err := hp.pushAggregation(ctx, plan, grouping, aggrs, false)
+	newPlan, groupingOffsets, aggrParamOffsets, pushed, err := hp.pushAggregation(ctx, plan, grouping, aggrs, false)
 	if err != nil {
 		return nil, err
 	}
+	if !pushed {
+		oa.preProcess = true
+		oa.aggrOnEngine = true
+	}
 
 	plan = newPlan
 
@@ -589,7 +332,7 @@ func (hp *horizonPlanning) planAggrUsingOA(
 		aggPlan = proj
 	}
 
-	aggrParams, err := generateAggregateParams(aggrs, aggrParamOffsets, proj)
+	aggrParams, err := generateAggregateParams(aggrs, aggrParamOffsets, proj, pushed)
 	if err != nil {
 		return nil, err
 	}
@@ -637,7 +380,7 @@ func passGroupingColumns(proj *projection, groupings []offsets, grouping []abstr
 	return projGrpOffsets, nil
 }
 
-func generateAggregateParams(aggrs []abstract.Aggr, aggrParamOffsets [][]offsets, proj *projection) ([]*engine.AggregateParams, error) {
+func generateAggregateParams(aggrs []abstract.Aggr, aggrParamOffsets [][]offsets, proj *projection, pushed bool) ([]*engine.AggregateParams, error) {
 	aggrParams := make([]*engine.AggregateParams, len(aggrs))
 	for idx, paramOffset := range aggrParamOffsets {
 		aggr := aggrs[idx]
@@ -668,10 +411,13 @@ func generateAggregateParams(aggrs []abstract.Aggr, aggrParamOffsets [][]offsets
 		}
 
 		opcode := engine.AggregateSum
-		if aggr.OpCode == engine.AggregateMin ||
-			aggr.OpCode == engine.AggregateMax ||
-			aggr.OpCode == engine.AggregateRandom {
+		switch aggr.OpCode {
+		case engine.AggregateMin, engine.AggregateMax, engine.AggregateRandom:
 			opcode = aggr.OpCode
+		case engine.AggregateCount, engine.AggregateCountStar, engine.AggregateCountDistinct, engine.AggregateSumDistinct:
+			if !pushed {
+				opcode = aggr.OpCode
+			}
 		}
 
 		aggrParams[idx] = &engine.AggregateParams{
@@ -1328,7 +1074,7 @@ func (hp *horizonPlanning) needDistinctHandling(
 	if !funcExpr.Distinct {
 		return false, nil, nil
 	}
-	if opcode != engine.AggregateCount && opcode != engine.AggregateSum {
+	if opcode != engine.AggregateCount && opcode != engine.AggregateSum && opcode != engine.AggregateCountStar {
 		return false, nil, nil
 	}
 	innerAliased, ok := funcExpr.Exprs[0].(*sqlparser.AliasedExpr)
diff --git a/go/vt/vtgate/planbuilder/ordered_aggregate.go b/go/vt/vtgate/planbuilder/ordered_aggregate.go
index c79120a8a8..fc1c9d155f 100644
--- a/go/vt/vtgate/planbuilder/ordered_aggregate.go
+++ b/go/vt/vtgate/planbuilder/ordered_aggregate.go
@@ -67,6 +67,8 @@ type orderedAggregate struct {
 	// preProcess is true if one of the aggregates needs preprocessing.
 	preProcess bool
 
+	aggrOnEngine bool
+
 	// aggregates specifies the aggregation parameters for each
 	// aggregation function: function opcode and input column number.
 	aggregates []*engine.AggregateParams
@@ -241,6 +243,7 @@ func (oa *orderedAggregate) Primitive() engine.Primitive {
 	if len(oa.groupByKeys) == 0 {
 		return &engine.ScalarAggregate{
 			PreProcess:          oa.preProcess,
+			AggrOnEngine:        oa.aggrOnEngine,
 			Aggregates:          oa.aggregates,
 			TruncateColumnCount: oa.truncateColumnCount,
 			Collations:          colls,
@@ -250,6 +253,7 @@ func (oa *orderedAggregate) Primitive() engine.Primitive {
 
 	return &engine.OrderedAggregate{
 		PreProcess:          oa.preProcess,
+		AggrOnEngine:        oa.aggrOnEngine,
 		Aggregates:          oa.aggregates,
 		GroupByKeys:         oa.groupByKeys,
 		TruncateColumnCount: oa.truncateColumnCount,
@@ -322,7 +326,7 @@ func (oa *orderedAggregate) needDistinctHandling(pb *primitiveBuilder, funcExpr
 	if !funcExpr.Distinct {
 		return false, nil, nil
 	}
-	if opcode != engine.AggregateCount && opcode != engine.AggregateSum {
+	if opcode != engine.AggregateCount && opcode != engine.AggregateSum && opcode != engine.AggregateCountStar {
 		return false, nil, nil
 	}
 	innerAliased, ok := funcExpr.Exprs[0].(*sqlparser.AliasedExpr)
@@ -364,6 +368,16 @@ func (oa *orderedAggregate) Wireup(plan logicalPlan, jt *jointab) error {
 			oa.truncateColumnCount = len(oa.resultColumns)
 		}
 	}
+	for _, key := range oa.aggregates {
+		switch key.Opcode {
+		case engine.AggregateCount:
+			if key.Alias == "" {
+				key.Alias = key.Opcode.String()
+			}
+			key.Opcode = engine.AggregateSum
+		}
+	}
+
 	return oa.input.Wireup(plan, jt)
 }
 
diff --git a/go/vt/vtgate/planbuilder/projection_pushing.go b/go/vt/vtgate/planbuilder/projection_pushing.go
new file mode 100644
index 0000000000..0cfd0dac47
--- /dev/null
+++ b/go/vt/vtgate/planbuilder/projection_pushing.go
@@ -0,0 +1,336 @@
+/*
+Copyright 2022 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 (
+	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/planbuilder/physical"
+	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
+	"vitess.io/vitess/go/vt/vtgate/semantics"
+)
+
+// pushProjection pushes a projection to the plan.
+func pushProjection(
+	ctx *plancontext.PlanningContext,
+	expr *sqlparser.AliasedExpr,
+	plan logicalPlan,
+	inner, reuseCol, hasAggregation bool,
+) (offset int, added bool, err error) {
+	switch node := plan.(type) {
+	case *limit, *projection, *pulloutSubquery, *distinct, *filter:
+		// All of these either push to the single source, or push to the LHS
+		src := node.Inputs()[0]
+		return pushProjection(ctx, expr, src, inner, reuseCol, hasAggregation)
+	case *routeGen4:
+		return addExpressionToRoute(ctx, node, expr, reuseCol)
+	case *hashJoin:
+		return pushProjectionIntoHashJoin(ctx, expr, node, reuseCol, inner, hasAggregation)
+	case *joinGen4:
+		return pushProjectionIntoJoin(ctx, expr, node, reuseCol, inner, hasAggregation)
+	case *simpleProjection:
+		return pushProjectionIntoSimpleProj(ctx, expr, node, inner, hasAggregation, reuseCol)
+	case *orderedAggregate:
+		return pushProjectionIntoOA(ctx, expr, node, inner, hasAggregation)
+	case *vindexFunc:
+		return pushProjectionIntoVindexFunc(node, expr, reuseCol)
+	case *semiJoin:
+		return pushProjectionIntoSemiJoin(ctx, expr, reuseCol, node, inner, hasAggregation)
+	case *concatenateGen4:
+		return pushProjectionIntoConcatenate(ctx, expr, hasAggregation, node, inner, reuseCol)
+	default:
+		return 0, false, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "[BUG] push projection does not yet support: %T", node)
+	}
+}
+
+func pushProjectionIntoVindexFunc(node *vindexFunc, expr *sqlparser.AliasedExpr, reuseCol bool) (int, bool, error) {
+	colsBefore := len(node.eVindexFunc.Cols)
+	i, err := node.SupplyProjection(expr, reuseCol)
+	if err != nil {
+		return 0, false, err
+	}
+	return i /* col added */, len(node.eVindexFunc.Cols) > colsBefore, nil
+}
+
+func pushProjectionIntoConcatenate(ctx *plancontext.PlanningContext, expr *sqlparser.AliasedExpr, hasAggregation bool, node *concatenateGen4, inner bool, reuseCol bool) (int, bool, error) {
+	if hasAggregation {
+		return 0, false, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: aggregation on unions")
+	}
+	offset, added, err := pushProjection(ctx, expr, node.sources[0], inner, reuseCol, hasAggregation)
+	if err != nil {
+		return 0, false, err
+	}
+	if added && ctx.SemTable.DirectDeps(expr.Expr).NumberOfTables() > 0 {
+		return 0, false, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "pushing projection %v on concatenate should reference an existing column", sqlparser.String(expr))
+	}
+	if added {
+		for _, source := range node.sources[1:] {
+			_, _, err := pushProjection(ctx, expr, source, inner, reuseCol, hasAggregation)
+			if err != nil {
+				return 0, false, err
+			}
+		}
+	}
+	return offset, added, nil
+}
+
+func pushProjectionIntoSemiJoin(
+	ctx *plancontext.PlanningContext,
+	expr *sqlparser.AliasedExpr,
+	reuseCol bool,
+	node *semiJoin,
+	inner, hasAggregation bool,
+) (int, bool, error) {
+	passDownReuseCol := reuseCol
+	if !reuseCol {
+		passDownReuseCol = expr.As.IsEmpty()
+	}
+	offset, added, err := pushProjection(ctx, expr, node.lhs, inner, passDownReuseCol, hasAggregation)
+	if err != nil {
+		return 0, false, err
+	}
+	column := -(offset + 1)
+	if reuseCol && !added {
+		for idx, col := range node.cols {
+			if column == col {
+				return idx, false, nil
+			}
+		}
+	}
+	node.cols = append(node.cols, column)
+	return len(node.cols) - 1, true, nil
+}
+
+func pushProjectionIntoOA(ctx *plancontext.PlanningContext, expr *sqlparser.AliasedExpr, node *orderedAggregate, inner, hasAggregation bool) (int, bool, error) {
+	colName, isColName := expr.Expr.(*sqlparser.ColName)
+	for _, aggregate := range node.aggregates {
+		if sqlparser.EqualsExpr(aggregate.Expr, expr.Expr) {
+			return aggregate.Col, false, nil
+		}
+		if isColName && colName.Name.EqualString(aggregate.Alias) {
+			return aggregate.Col, false, nil
+		}
+	}
+	for _, key := range node.groupByKeys {
+		if sqlparser.EqualsExpr(key.Expr, expr.Expr) {
+			return key.KeyCol, false, nil
+		}
+	}
+	offset, _, err := pushProjection(ctx, expr, node.input, inner, true, hasAggregation)
+	if err != nil {
+		return 0, false, err
+	}
+	node.aggregates = append(node.aggregates, &engine.AggregateParams{
+		Opcode:   engine.AggregateRandom,
+		Col:      offset,
+		Alias:    expr.ColumnName(),
+		Expr:     expr.Expr,
+		Original: expr,
+	})
+	return offset, true, nil
+}
+
+func pushProjectionIntoSimpleProj(
+	ctx *plancontext.PlanningContext,
+	expr *sqlparser.AliasedExpr,
+	node *simpleProjection,
+	inner, hasAggregation, reuseCol bool,
+) (int, bool, error) {
+	offset, _, err := pushProjection(ctx, expr, node.input, inner, true, hasAggregation)
+	if err != nil {
+		return 0, false, err
+	}
+	for i, value := range node.eSimpleProj.Cols {
+		// we return early if we already have the column in the simple projection's
+		// output list so we do not add it again.
+		if reuseCol && value == offset {
+			return i, false, nil
+		}
+	}
+	node.eSimpleProj.Cols = append(node.eSimpleProj.Cols, offset)
+	return len(node.eSimpleProj.Cols) - 1, true, nil
+}
+
+func pushProjectionIntoJoin(
+	ctx *plancontext.PlanningContext,
+	expr *sqlparser.AliasedExpr,
+	node *joinGen4,
+	reuseCol, inner, hasAggregation bool,
+) (int, bool, error) {
+	lhsSolves := node.Left.ContainsTables()
+	rhsSolves := node.Right.ContainsTables()
+	deps := ctx.SemTable.RecursiveDeps(expr.Expr)
+	var column int
+	var appended bool
+	passDownReuseCol := reuseCol
+	if !reuseCol {
+		passDownReuseCol = expr.As.IsEmpty()
+	}
+	switch {
+	case deps.IsSolvedBy(lhsSolves):
+		offset, added, err := pushProjection(ctx, expr, node.Left, inner, passDownReuseCol, hasAggregation)
+		if err != nil {
+			return 0, false, err
+		}
+		column = -(offset + 1)
+		appended = added
+	case deps.IsSolvedBy(rhsSolves):
+		offset, added, err := pushProjection(ctx, expr, node.Right, inner && node.Opcode != engine.LeftJoin, passDownReuseCol, hasAggregation)
+		if err != nil {
+			return 0, false, err
+		}
+		column = offset + 1
+		appended = added
+	default:
+		// if an expression has aggregation, then it should not be split up and pushed to both sides,
+		// for example an expression like count(*) will have dependencies on both sides, but we should not push it
+		// instead we should return an error
+		if hasAggregation {
+			return 0, false, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: cross-shard query with aggregates")
+		}
+		// now we break the expression into left and right side dependencies and rewrite the left ones to bind variables
+		bvName, cols, rewrittenExpr, err := physical.BreakExpressionInLHSandRHS(ctx, expr.Expr, lhsSolves)
+		if err != nil {
+			return 0, false, err
+		}
+		// go over all the columns coming from the left side of the tree and push them down. While at it, also update the bind variable map.
+		// It is okay to reuse the columns on the left side since
+		// the final expression which will be selected will be pushed into the right side.
+		for i, col := range cols {
+			colOffset, _, err := pushProjection(ctx, &sqlparser.AliasedExpr{Expr: col}, node.Left, inner, true, false)
+			if err != nil {
+				return 0, false, err
+			}
+			node.Vars[bvName[i]] = colOffset
+		}
+		// push the rewritten expression on the right side of the tree. Here we should take care whether we want to reuse the expression or not.
+		expr.Expr = rewrittenExpr
+		offset, added, err := pushProjection(ctx, expr, node.Right, inner && node.Opcode != engine.LeftJoin, passDownReuseCol, false)
+		if err != nil {
+			return 0, false, err
+		}
+		column = offset + 1
+		appended = added
+	}
+	if reuseCol && !appended {
+		for idx, col := range node.Cols {
+			if column == col {
+				return idx, false, nil
+			}
+		}
+		// the column was not appended to either child, but we could not find it in out cols list,
+		// so we'll still add it
+	}
+	node.Cols = append(node.Cols, column)
+	return len(node.Cols) - 1, true, nil
+}
+
+func pushProjectionIntoHashJoin(
+	ctx *plancontext.PlanningContext,
+	expr *sqlparser.AliasedExpr,
+	node *hashJoin,
+	reuseCol, inner, hasAggregation bool,
+) (int, bool, error) {
+	lhsSolves := node.Left.ContainsTables()
+	rhsSolves := node.Right.ContainsTables()
+	deps := ctx.SemTable.RecursiveDeps(expr.Expr)
+	var column int
+	var appended bool
+	passDownReuseCol := reuseCol
+	if !reuseCol {
+		passDownReuseCol = expr.As.IsEmpty()
+	}
+	switch {
+	case deps.IsSolvedBy(lhsSolves):
+		offset, added, err := pushProjection(ctx, expr, node.Left, inner, passDownReuseCol, hasAggregation)
+		if err != nil {
+			return 0, false, err
+		}
+		column = -(offset + 1)
+		appended = added
+	case deps.IsSolvedBy(rhsSolves):
+		offset, added, err := pushProjection(ctx, expr, node.Right, inner && node.Opcode != engine.LeftJoin, passDownReuseCol, hasAggregation)
+		if err != nil {
+			return 0, false, err
+		}
+		column = offset + 1
+		appended = added
+	default:
+		// if an expression has aggregation, then it should not be split up and pushed to both sides,
+		// for example an expression like count(*) will have dependencies on both sides, but we should not push it
+		// instead we should return an error
+		if hasAggregation {
+			return 0, false, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: cross-shard query with aggregates")
+		}
+		return 0, false, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: hash join with projection from both sides of the join")
+	}
+	if reuseCol && !appended {
+		for idx, col := range node.Cols {
+			if column == col {
+				return idx, false, nil
+			}
+		}
+		// the column was not appended to either child, but we could not find it in out cols list,
+		// so we'll still add it
+	}
+	node.Cols = append(node.Cols, column)
+	return len(node.Cols) - 1, true, nil
+}
+
+func addExpressionToRoute(ctx *plancontext.PlanningContext, rb *routeGen4, expr *sqlparser.AliasedExpr, reuseCol bool) (int, bool, error) {
+	if reuseCol {
+		if i := checkIfAlreadyExists(expr, rb.Select, ctx.SemTable); i != -1 {
+			return i, false, nil
+		}
+	}
+	expr.Expr = sqlparser.RemoveKeyspaceFromColName(expr.Expr)
+	sel, isSel := rb.Select.(*sqlparser.Select)
+	if !isSel {
+		return 0, false, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.BadFieldError, "unsupported: pushing projection '%s' on %T", sqlparser.String(expr), rb.Select)
+	}
+
+	if ctx.RewriteDerivedExpr {
+		// if we are trying to push a projection that belongs to a DerivedTable
+		// we rewrite that expression, so it matches the column name used inside
+		// that derived table.
+		err := rewriteProjectionOfDerivedTable(expr, ctx.SemTable)
+		if err != nil {
+			return 0, false, err
+		}
+	}
+
+	offset := len(sel.SelectExprs)
+	sel.SelectExprs = append(sel.SelectExprs, expr)
+	return offset, true, nil
+}
+
+func rewriteProjectionOfDerivedTable(expr *sqlparser.AliasedExpr, semTable *semantics.SemTable) error {
+	ti, err := semTable.TableInfoForExpr(expr.Expr)
+	if err != nil && err != semantics.ErrMultipleTables {
+		return err
+	}
+	_, isDerivedTable := ti.(*semantics.DerivedTable)
+	if isDerivedTable {
+		expr.Expr, err = semantics.RewriteDerivedExpression(expr.Expr, ti)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/go/vt/wrangler/vdiff.go b/go/vt/wrangler/vdiff.go
index 32b18037fc..98a8152a20 100644
--- a/go/vt/wrangler/vdiff.go
+++ b/go/vt/wrangler/vdiff.go
@@ -565,8 +565,13 @@ func (df *vdiff) buildTablePlan(table *tabletmanagerdatapb.TableDefinition, quer
 			if expr, ok := selExpr.Expr.(*sqlparser.FuncExpr); ok {
 				switch fname := expr.Name.Lowered(); fname {
 				case "count", "sum":
+					// this will only work as long as aggregates can be pushed down to tablets
+					// this won't work: "select count(*) from (select id from t limit 1)"
+					// since vreplication only handles simple tables (no joins/derived tables) this is fine for now
+					// but will need to be revisited when we add such support to vreplication
+					aggregateFuncType := "sum"
 					aggregates = append(aggregates, &engine.AggregateParams{
-						Opcode: engine.SupportedAggregates[fname],
+						Opcode: engine.SupportedAggregates[aggregateFuncType],
 						Col:    len(sourceSelect.SelectExprs) - 1,
 					})
 				}


Test patch:
diff --git a/go/test/endtoend/utils/cmp.go b/go/test/endtoend/utils/cmp.go
index 70dc42be8a..21146f1ac2 100644
--- a/go/test/endtoend/utils/cmp.go
+++ b/go/test/endtoend/utils/cmp.go
@@ -120,6 +120,22 @@ func (mcmp *MySQLCompare) AssertFoundRowsValue(query, workload string, count int
 	assert.Equalf(mcmp.t, want, got, "Workload: %s\nQuery:%s\n", workload, query)
 }
 
+// AssertMatchesNoCompare compares the record of mysql and vitess separately and not with each other.
+func (mcmp *MySQLCompare) AssertMatchesNoCompare(query, mExp string, vExp string) {
+	mcmp.t.Helper()
+	mQr, vQr := mcmp.execNoCompare(query)
+	got := fmt.Sprintf("%v", mQr.Rows)
+	diff := cmp.Diff(mExp, got)
+	if diff != "" {
+		mcmp.t.Errorf("MySQL Query: %s (-want +got):\n%s\nGot:%s", query, diff, got)
+	}
+	got = fmt.Sprintf("%v", vQr.Rows)
+	diff = cmp.Diff(vExp, got)
+	if diff != "" {
+		mcmp.t.Errorf("Vitess Query: %s (-want +got):\n%s\nGot:%s", query, diff, got)
+	}
+}
+
 // Exec executes the given query against both Vitess and MySQL and compares
 // the two result set. If there is a mismatch, the difference will be printed and the
 // test will fail. If the query produces an error in either Vitess or MySQL, the test
@@ -136,6 +152,16 @@ func (mcmp *MySQLCompare) Exec(query string) *sqltypes.Result {
 	return vtQr
 }
 
+func (mcmp *MySQLCompare) execNoCompare(query string) (*sqltypes.Result, *sqltypes.Result) {
+	mcmp.t.Helper()
+	vtQr, err := mcmp.VtConn.ExecuteFetch(query, 1000, true)
+	require.NoError(mcmp.t, err, "[Vitess Error] for query: "+query)
+
+	mysqlQr, err := mcmp.MySQLConn.ExecuteFetch(query, 1000, true)
+	require.NoError(mcmp.t, err, "[MySQL Error] for query: "+query)
+	return mysqlQr, vtQr
+}
+
 // ExecWithColumnCompare executes the given query against both Vitess and MySQL and compares
 // the two result set. If there is a mismatch, the difference will be printed and the
 // test will fail. If the query produces an error in either Vitess or MySQL, the test
diff --git a/go/test/endtoend/vtgate/queries/aggregation/aggregation_test.go b/go/test/endtoend/vtgate/queries/aggregation/aggregation_test.go
index 0e20922288..fbc307fe5a 100644
--- a/go/test/endtoend/vtgate/queries/aggregation/aggregation_test.go
+++ b/go/test/endtoend/vtgate/queries/aggregation/aggregation_test.go
@@ -17,13 +17,11 @@ limitations under the License.
 package aggregation
 
 import (
-	"context"
 	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 
-	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/test/endtoend/cluster"
 	"vitess.io/vitess/go/test/endtoend/utils"
 )
@@ -303,18 +301,42 @@ func TestGreaterEqualFilterOnScatter(t *testing.T) {
 }
 
 func TestGroupByOnlyFullGroupByOff(t *testing.T) {
-	vtParams := mysql.ConnParams{
-		Host: "localhost",
-		Port: clusterInstance.VtgateMySQLPort,
-	}
-	conn, err := mysql.Connect(context.Background(), &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
+	mcmp, closer := start(t)
+	defer closer()
 
-	utils.Exec(t, conn, "insert into t9(id1, id2, id3) values(1,'a', '1'), (2,'Abc','2'), (3,'b', '3'), (4,'c', '4'), (5,'test', '5')")
-	utils.Exec(t, conn, "insert into t9(id1, id2, id3) values(6,'a', '11'), (7,'Abc','22'), (8,'b', '33'), (9,'c', '44'), (10,'test', '55')")
-	utils.Exec(t, conn, "set @@sql_mode = ' '")
+	mcmp.Exec("insert into t9(id1, id2, id3) values(1,'a', '1'), (2,'Abc','2'), (3,'b', '3'), (4,'c', '4'), (5,'test', '5')")
+	mcmp.Exec("insert into t9(id1, id2, id3) values(6,'a', '11'), (7,'Abc','22'), (8,'b', '33'), (9,'c', '44'), (10,'test', '55')")
+	mcmp.Exec("set @@sql_mode = ' '")
 
 	// We do not use AssertMatches here because the results for the second column are random
-	utils.Exec(t, conn, "select /*vt+ PLANNER=gen4 */ id2, id3 from t9 group by id2")
+	_, err := mcmp.ExecAndIgnore("select /*vt+ PLANNER=gen4 */ id2, id3 from t9 group by id2")
+	require.NoError(t, err)
+}
+
+func TestAggOnTopOfLimit(t *testing.T) {
+	mcmp, closer := start(t)
+	defer closer()
+	mcmp.Exec("insert into aggr_test(id, val1, val2) values(1,'a',6), (2,'a',1), (3,'b',1), (4,'c',3), (5,'c',4), (6,'b',null), (7,null,2), (8,null,null)")
+
+	for _, workload := range []string{"oltp", "olap"} {
+		t.Run(workload, func(t *testing.T) {
+			utils.Exec(t, mcmp.VtConn, fmt.Sprintf("set workload = '%s'", workload))
+			mcmp.AssertMatches(" select /*vt+ PLANNER=gen4 */ count(*) from (select id, val1 from aggr_test where val2 < 4 limit 2) as x", "[[INT64(2)]]")
+			mcmp.AssertMatches(" select /*vt+ PLANNER=gen4 */ count(val1) from (select id, val1 from aggr_test where val2 < 4 order by val1 desc limit 2) as x", "[[INT64(2)]]")
+			mcmp.AssertMatches(" select /*vt+ PLANNER=gen4 */ count(*) from (select id, val1 from aggr_test where val2 is null limit 2) as x", "[[INT64(2)]]")
+			mcmp.AssertMatches(" select /*vt+ PLANNER=gen4 */ count(val1) from (select id, val1 from aggr_test where val2 is null limit 2) as x", "[[INT64(1)]]")
+			mcmp.AssertMatches(" select /*vt+ PLANNER=gen4 */ count(val2) from (select id, val2 from aggr_test where val2 is null limit 2) as x", "[[INT64(0)]]")
+			mcmp.AssertMatches(" select /*vt+ PLANNER=gen4 */ val1, count(*) from (select id, val1 from aggr_test where val2 < 4 order by val1 limit 2) as x group by val1", `[[NULL INT64(1)] [VARCHAR("a") INT64(1)]]`)
+			mcmp.AssertMatchesNoOrder(" select /*vt+ PLANNER=gen4 */ val1, count(val2) from (select val1, val2 from aggr_test limit 8) as x group by val1", `[[NULL INT64(1)] [VARCHAR("a") INT64(2)] [VARCHAR("b") INT64(1)] [VARCHAR("c") INT64(2)]]`)
+
+			// mysql returns FLOAT64(0), vitess returns DECIMAL(0)
+			mcmp.AssertMatchesNoCompare(" select /*vt+ PLANNER=gen4 */ count(*), sum(val1) from (select id, val1 from aggr_test where val2 < 4 order by val1 desc limit 2) as x", "[[INT64(2) FLOAT64(0)]]", "[[INT64(2) DECIMAL(0)]]")
+			mcmp.AssertMatches(" select /*vt+ PLANNER=gen4 */ count(val1), sum(id) from (select id, val1 from aggr_test where val2 < 4 order by val1 desc limit 2) as x", "[[INT64(2) DECIMAL(7)]]")
+			mcmp.AssertMatches(" select /*vt+ PLANNER=gen4 */ count(*), sum(id) from (select id, val1 from aggr_test where val2 is null limit 2) as x", "[[INT64(2) DECIMAL(14)]]")
+			mcmp.AssertMatches(" select /*vt+ PLANNER=gen4 */ count(val1), sum(id) from (select id, val1 from aggr_test where val2 is null limit 2) as x", "[[INT64(1) DECIMAL(14)]]")
+			mcmp.AssertMatches(" select /*vt+ PLANNER=gen4 */ count(val2), sum(val2) from (select id, val2 from aggr_test where val2 is null limit 2) as x", "[[INT64(0) NULL]]")
+			mcmp.AssertMatches(" select /*vt+ PLANNER=gen4 */ val1, count(*), sum(id) from (select id, val1 from aggr_test where val2 < 4 order by val1 limit 2) as x group by val1", `[[NULL INT64(1) DECIMAL(7)] [VARCHAR("a") INT64(1) DECIMAL(2)]]`)
+			mcmp.AssertMatchesNoOrder(" select /*vt+ PLANNER=gen4 */ val1, count(val2), sum(val2) from (select val1, val2 from aggr_test limit 8) as x group by val1", `[[NULL INT64(1) DECIMAL(2)] [VARCHAR("a") INT64(2) DECIMAL(7)] [VARCHAR("b") INT64(1) DECIMAL(1)] [VARCHAR("c") INT64(2) DECIMAL(7)]]`)
+		})
+	}
 }
diff --git a/go/vt/vtgate/engine/ordered_aggregate_test.go b/go/vt/vtgate/engine/ordered_aggregate_test.go
index e8fcc0cac4..3f816a57f3 100644
--- a/go/vt/vtgate/engine/ordered_aggregate_test.go
+++ b/go/vt/vtgate/engine/ordered_aggregate_test.go
@@ -51,7 +51,7 @@ func TestOrderedAggregateExecute(t *testing.T) {
 
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    1,
 		}},
 		GroupByKeys: []*GroupByParams{{KeyCol: 0}},
@@ -88,7 +88,7 @@ func TestOrderedAggregateExecuteTruncate(t *testing.T) {
 
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    1,
 		}},
 		GroupByKeys:         []*GroupByParams{{KeyCol: 2}},
@@ -130,7 +130,7 @@ func TestOrderedAggregateStreamExecute(t *testing.T) {
 
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    1,
 		}},
 		GroupByKeys: []*GroupByParams{{KeyCol: 0}},
@@ -173,7 +173,7 @@ func TestOrderedAggregateStreamExecuteTruncate(t *testing.T) {
 
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    1,
 		}},
 		GroupByKeys:         []*GroupByParams{{KeyCol: 2}},
@@ -314,7 +314,7 @@ func TestOrderedAggregateExecuteCountDistinct(t *testing.T) {
 			Alias:  "count(distinct col2)",
 		}, {
 			// Also add a count(*)
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    2,
 		}},
 		GroupByKeys: []*GroupByParams{{KeyCol: 0}},
@@ -390,7 +390,7 @@ func TestOrderedAggregateStreamCountDistinct(t *testing.T) {
 			Alias:  "count(distinct col2)",
 		}, {
 			// Also add a count(*)
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    2,
 		}},
 		GroupByKeys: []*GroupByParams{{KeyCol: 0}},
@@ -561,7 +561,7 @@ func TestOrderedAggregateKeysFail(t *testing.T) {
 
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    1,
 		}},
 		GroupByKeys: []*GroupByParams{{KeyCol: 0}},
@@ -594,7 +594,7 @@ func TestOrderedAggregateMergeFail(t *testing.T) {
 
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    1,
 		}},
 		GroupByKeys: []*GroupByParams{{KeyCol: 0}},
@@ -634,7 +634,7 @@ func TestMerge(t *testing.T) {
 	assert := assert.New(t)
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    1,
 		}, {
 			Opcode: AggregateSum,
diff --git a/go/vt/vtgate/engine/scalar_aggregation_test.go b/go/vt/vtgate/engine/scalar_aggregation_test.go
index 0e8987e4d4..0652d7679e 100644
--- a/go/vt/vtgate/engine/scalar_aggregation_test.go
+++ b/go/vt/vtgate/engine/scalar_aggregation_test.go
@@ -113,7 +113,7 @@ func TestScalarAggregateStreamExecute(t *testing.T) {
 
 	oa := &ScalarAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    0,
 		}},
 		Input:               fp,
diff --git a/go/vt/vtgate/executor_framework_test.go b/go/vt/vtgate/executor_framework_test.go
index 30e6ff9888..c699ccc7a5 100644
--- a/go/vt/vtgate/executor_framework_test.go
+++ b/go/vt/vtgate/executor_framework_test.go
@@ -140,15 +140,7 @@ var executorVSchema = `
 			"params": {
 				"region_bytes": "1"
 			}
-    	},
-		"multicol_vdx": {
-			"type": "multicol",
-			"params": {
-				"column_count": "3",
-				"column_bytes": "1,3,4",
-				"column_vindex": "hash,binary,unicode_loose_xxhash"
-			}
-        }
+    	}
 	},
 	"tables": {
 		"user": {
@@ -328,15 +320,7 @@ var executorVSchema = `
 					"name": "regional_vdx"
 				}
 			]
-    	},
-		"multicoltbl": {
-			"column_vindexes": [
-				{
-					"columns": ["cola","colb","colc"],
-					"name": "multicol_vdx"
-				}
-			]
-		}
+    	}
 	}
 }
 `
@@ -443,7 +427,7 @@ func init() {
 func createExecutorEnv() (executor *Executor, sbc1, sbc2, sbclookup *sandboxconn.SandboxConn) {
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	serv := newSandboxForCells([]string{cell})
 	resolver := newTestResolver(hc, serv, cell)
@@ -498,7 +482,7 @@ func createExecutorEnv() (executor *Executor, sbc1, sbc2, sbclookup *sandboxconn
 func createCustomExecutor(vschema string) (executor *Executor, sbc1, sbc2, sbclookup *sandboxconn.SandboxConn) {
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = vschema
 	serv := newSandboxForCells([]string{cell})
 	resolver := newTestResolver(hc, serv, cell)
@@ -520,7 +504,7 @@ func createCustomExecutor(vschema string) (executor *Executor, sbc1, sbc2, sbclo
 func createCustomExecutorSetValues(vschema string, values []*sqltypes.Result) (executor *Executor, sbc1, sbc2, sbclookup *sandboxconn.SandboxConn) {
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = vschema
 	serv := newSandboxForCells([]string{cell})
 	resolver := newTestResolver(hc, serv, cell)
diff --git a/go/vt/vtgate/executor_select_test.go b/go/vt/vtgate/executor_select_test.go
index 9b5e1277a2..d61e23b47e 100644
--- a/go/vt/vtgate/executor_select_test.go
+++ b/go/vt/vtgate/executor_select_test.go
@@ -1494,7 +1494,7 @@ func TestSelectScatter(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -1529,7 +1529,7 @@ func TestSelectScatterPartial(t *testing.T) {
 	}
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -1589,7 +1589,7 @@ func TestSelectScatterPartialOLAP(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -1640,7 +1640,7 @@ func TestSelectScatterPartialOLAP2(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -1696,7 +1696,7 @@ func TestStreamSelectScatter(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -1731,7 +1731,7 @@ func TestSelectScatterOrderBy(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -1797,7 +1797,7 @@ func TestSelectScatterOrderByVarChar(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -1861,7 +1861,7 @@ func TestStreamSelectScatterOrderBy(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -1919,7 +1919,7 @@ func TestStreamSelectScatterOrderByVarChar(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -1977,7 +1977,7 @@ func TestSelectScatterAggregate(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -2036,7 +2036,7 @@ func TestStreamSelectScatterAggregate(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -2096,7 +2096,7 @@ func TestSelectScatterLimit(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -2164,7 +2164,7 @@ func TestStreamSelectScatterLimit(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -2963,7 +2963,7 @@ func TestStreamOrderByLimitWithMultipleResults(t *testing.T) {
 	// Special setup: Don't use createExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -2997,7 +2997,7 @@ func TestSelectScatterFails(t *testing.T) {
 	sess := &vtgatepb.Session{}
 	cell := "aa"
 	hc := discovery.NewFakeHealthCheck(nil)
-	s := createSandbox("TestExecutor")
+	s := createSandbox(KsTestSharded)
 	s.VSchema = executorVSchema
 	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
 	serv := new(sandboxTopo)
@@ -3364,15 +3364,12 @@ func TestMultiCol(t *testing.T) {
 	// Special setup: Don't use createLegacyExecutorEnv.
 
 	*plannerVersion = "gen4"
-	defer func() {
-		*plannerVersion = "v3"
-	}()
 	cell := "multicol"
-	ks := "TestExecutor"
+	ks := "TestMultiCol"
 	hc := discovery.NewFakeHealthCheck(nil)
 	s := createSandbox(ks)
 	s.ShardSpec = "-20-20a0-"
-	s.VSchema = executorVSchema
+	s.VSchema = multiColVschema
 	serv := newSandboxForCells([]string{cell})
 	resolver := newTestResolver(hc, serv, cell)
 	shards := []string{"-20", "20-20a0", "20a0-"}
@@ -3417,19 +3414,41 @@ func TestMultiCol(t *testing.T) {
 	}
 }
 
+var multiColVschema = `
+{
+	"sharded": true,
+	"vindexes": {
+		"multicol_vdx": {
+			"type": "multicol",
+			"params": {
+				"column_count": "3",
+				"column_bytes": "1,3,4",
+				"column_vindex": "hash,binary,unicode_loose_xxhash"
+			}
+        }
+	},
+	"tables": {
+		"multicoltbl": {
+			"column_vindexes": [
+				{
+					"columns": ["cola","colb","colc"],
+					"name": "multicol_vdx"
+				}
+			]
+		}
+	}
+}
+`
+
 func TestMultiColPartial(t *testing.T) {
 	// Special setup: Don't use createLegacyExecutorEnv.
-
 	*plannerVersion = "gen4"
-	defer func() {
-		*plannerVersion = "v3"
-	}()
 	cell := "multicol"
-	ks := "TestExecutor"
+	ks := "TestMultiCol"
 	hc := discovery.NewFakeHealthCheck(nil)
 	s := createSandbox(ks)
 	s.ShardSpec = "-20-20a0c0-"
-	s.VSchema = executorVSchema
+	s.VSchema = multiColVschema
 	serv := newSandboxForCells([]string{cell})
 	resolver := newTestResolver(hc, serv, cell)
 	shards := []string{"-20", "20-20a0c0", "20a0c0-"}
@@ -3482,3 +3501,227 @@ func TestMultiColPartial(t *testing.T) {
 		})
 	}
 }
+
+func TestSelectAggregationNoData(t *testing.T) {
+	// Special setup: Don't use createExecutorEnv.
+	*plannerVersion = "gen4"
+	cell := "aa"
+	hc := discovery.NewFakeHealthCheck(nil)
+	createSandbox(KsTestSharded).VSchema = executorVSchema
+	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
+	serv := new(sandboxTopo)
+	resolver := newTestResolver(hc, serv, cell)
+	shards := []string{"-20", "20-40", "40-60", "60-80", "80-a0", "a0-c0", "c0-e0", "e0-"}
+	var conns []*sandboxconn.SandboxConn
+	for _, shard := range shards {
+		sbc := hc.AddTestTablet(cell, shard, 1, KsTestSharded, shard, topodatapb.TabletType_PRIMARY, true, 1, nil)
+		conns = append(conns, sbc)
+	}
+	executor := createExecutor(serv, cell, resolver)
+
+	tcases := []struct {
+		sql         string
+		sandboxRes  *sqltypes.Result
+		expSandboxQ string
+		expField    string
+		expRow      string
+	}{
+		{
+			sql:         `select count(distinct col) from user`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col", "int64")),
+			expSandboxQ: "select col, weight_string(col) from `user` group by col, weight_string(col) order by col asc",
+			expField:    `[name:"count(distinct col)" type:INT64]`,
+			expRow:      `[[INT64(0)]]`,
+		},
+		{
+			sql:         `select count(*) from user`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("count(*)", "int64"), "0"),
+			expSandboxQ: "select count(*) from `user`",
+			expField:    `[name:"count(*)" type:INT64]`,
+			expRow:      `[[INT64(0)]]`,
+		},
+		{
+			sql:         `select col, count(*) from user group by col`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col|count(*)", "int64|int64")),
+			expSandboxQ: "select col, count(*), weight_string(col) from `user` group by col, weight_string(col) order by col asc",
+			expField:    `[name:"col" type:INT64 name:"count(*)" type:INT64]`,
+			expRow:      `[]`,
+		},
+		{
+			sql:         `select col, count(*) from user group by col limit 2`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col|count(*)", "int64|int64")),
+			expSandboxQ: "select col, count(*), weight_string(col) from `user` group by col, weight_string(col) order by col asc limit :__upper_limit",
+			expField:    `[name:"col" type:INT64 name:"count(*)" type:INT64]`,
+			expRow:      `[]`,
+		},
+		{
+			sql:         `select count(*) from (select col1, col2 from user limit 2) x`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2", "int64|int64")),
+			expSandboxQ: "select col1, col2 from `user` limit :__upper_limit",
+			expField:    `[name:"count(*)" type:INT64]`,
+			expRow:      `[[INT64(0)]]`,
+		},
+		{
+			sql:         `select col2, count(*) from (select col1, col2 from user limit 2) x group by col2`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col2)", "int64|int64|varbinary")),
+			expSandboxQ: "select col1, col2, weight_string(col2) from `user` order by col2 asc limit :__upper_limit",
+			expField:    `[name:"col2" type:INT64 name:"count(*)" type:INT64]`,
+			expRow:      `[]`,
+		},
+	}
+
+	for _, tc := range tcases {
+		t.Run(tc.sql, func(t *testing.T) {
+			for _, sbc := range conns {
+				sbc.SetResults([]*sqltypes.Result{tc.sandboxRes})
+				sbc.Queries = nil
+			}
+			qr, err := executorExec(executor, tc.sql, nil)
+			require.NoError(t, err)
+			assert.Equal(t, tc.expField, fmt.Sprintf("%v", qr.Fields))
+			assert.Equal(t, tc.expRow, fmt.Sprintf("%v", qr.Rows))
+			require.Len(t, conns[0].Queries, 1)
+			assert.Equal(t, tc.expSandboxQ, conns[0].Queries[0].Sql)
+		})
+	}
+}
+
+func TestSelectAggregationData(t *testing.T) {
+	// Special setup: Don't use createExecutorEnv.
+	*plannerVersion = "gen4"
+	cell := "aa"
+	hc := discovery.NewFakeHealthCheck(nil)
+	createSandbox(KsTestSharded).VSchema = executorVSchema
+	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
+	serv := new(sandboxTopo)
+	resolver := newTestResolver(hc, serv, cell)
+	shards := []string{"-20", "20-40", "40-60", "60-80", "80-a0", "a0-c0", "c0-e0", "e0-"}
+	var conns []*sandboxconn.SandboxConn
+	for _, shard := range shards {
+		sbc := hc.AddTestTablet(cell, shard, 1, KsTestSharded, shard, topodatapb.TabletType_PRIMARY, true, 1, nil)
+		conns = append(conns, sbc)
+	}
+	executor := createExecutor(serv, cell, resolver)
+
+	tcases := []struct {
+		sql         string
+		sandboxRes  *sqltypes.Result
+		expSandboxQ string
+		expField    string
+		expRow      string
+	}{
+		{
+			sql:         `select count(distinct col) from user`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col", "int64"), "1", "2", "2", "3"),
+			expSandboxQ: "select col, weight_string(col) from `user` group by col, weight_string(col) order by col asc",
+			expField:    `[name:"count(distinct col)" type:INT64]`,
+			expRow:      `[[INT64(3)]]`,
+		},
+		{
+			sql:         `select count(*) from user`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("count(*)", "int64"), "3"),
+			expSandboxQ: "select count(*) from `user`",
+			expField:    `[name:"count(*)" type:INT64]`,
+			expRow:      `[[INT64(24)]]`,
+		},
+		{
+			sql:         `select col, count(*) from user group by col`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col|count(*)", "int64|int64"), "1|3"),
+			expSandboxQ: "select col, count(*), weight_string(col) from `user` group by col, weight_string(col) order by col asc",
+			expField:    `[name:"col" type:INT64 name:"count(*)" type:INT64]`,
+			expRow:      `[[INT64(1) INT64(24)]]`,
+		},
+		{
+			sql:         `select col, count(*) from user group by col limit 2`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col|count(*)", "int64|int64"), "1|2", "2|1", "3|4"),
+			expSandboxQ: "select col, count(*), weight_string(col) from `user` group by col, weight_string(col) order by col asc limit :__upper_limit",
+			expField:    `[name:"col" type:INT64 name:"count(*)" type:INT64]`,
+			expRow:      `[[INT64(1) INT64(16)] [INT64(2) INT64(8)]]`,
+		},
+		{
+			sql:         `select count(*) from (select col1, col2 from user limit 2) x`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2", "int64|int64"), "1|2", "2|1"),
+			expSandboxQ: "select col1, col2 from `user` limit :__upper_limit",
+			expField:    `[name:"count(*)" type:INT64]`,
+			expRow:      `[[INT64(2)]]`,
+		},
+		{
+			sql:         `select col2, count(*) from (select col1, col2 from user limit 9) x group by col2`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col2)", "int64|int64|varbinary"), "3|1|NULL", "2|2|NULL"),
+			expSandboxQ: "select col1, col2, weight_string(col2) from `user` order by col2 asc limit :__upper_limit",
+			expField:    `[name:"col2" type:INT64 name:"count(*)" type:INT64]`,
+			expRow:      `[[INT64(1) INT64(8)] [INT64(2) INT64(1)]]`,
+		},
+		{
+			sql:         `select count(col1) from (select id, col1 from user limit 2) x`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("id|col1", "int64|varchar"), "3|a", "2|b"),
+			expSandboxQ: "select id, col1 from `user` limit :__upper_limit",
+			expField:    `[name:"count(col1)" type:INT64]`,
+			expRow:      `[[INT64(2)]]`,
+		},
+		{
+			sql:         `select count(col1), col2 from (select col2, col1 from user limit 9) x group by col2`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col2|col1|weight_string(col2)", "int64|varchar|varbinary"), "3|a|NULL", "2|b|NULL"),
+			expSandboxQ: "select col2, col1, weight_string(col2) from `user` order by col2 asc limit :__upper_limit",
+			expField:    `[name:"count(col1)" type:INT64 name:"col2" type:INT64]`,
+			expRow:      `[[INT64(8) INT64(2)] [INT64(1) INT64(3)]]`,
+		},
+		{
+			sql:         `select col1, count(col2) from (select col1, col2 from user limit 9) x group by col1`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col1)", "varchar|int64|varbinary"), "a|1|a", "b|null|b"),
+			expSandboxQ: "select col1, col2, weight_string(col1) from `user` order by col1 asc limit :__upper_limit",
+			expField:    `[name:"col1" type:VARCHAR name:"count(col2)" type:INT64]`,
+			expRow:      `[[VARCHAR("a") INT64(8)] [VARCHAR("b") INT64(0)]]`,
+		},
+		{
+			sql:         `select col1, count(col2) from (select col1, col2 from user limit 32) x group by col1`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col1)", "varchar|int64|varbinary"), "null|1|null", "null|null|null", "a|1|a", "b|null|b"),
+			expSandboxQ: "select col1, col2, weight_string(col1) from `user` order by col1 asc limit :__upper_limit",
+			expField:    `[name:"col1" type:VARCHAR name:"count(col2)" type:INT64]`,
+			expRow:      `[[NULL INT64(8)] [VARCHAR("a") INT64(8)] [VARCHAR("b") INT64(0)]]`,
+		},
+		{
+			sql:         `select col1, sum(col2) from (select col1, col2 from user limit 4) x group by col1`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col1)", "varchar|int64|varbinary"), "a|3|a"),
+			expSandboxQ: "select col1, col2, weight_string(col1) from `user` order by col1 asc limit :__upper_limit",
+			expField:    `[name:"col1" type:VARCHAR name:"sum(col2)" type:DECIMAL]`,
+			expRow:      `[[VARCHAR("a") DECIMAL(12)]]`,
+		},
+		{
+			sql:         `select col1, sum(col2) from (select col1, col2 from user limit 4) x group by col1`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col1)", "varchar|varchar|varbinary"), "a|2|a"),
+			expSandboxQ: "select col1, col2, weight_string(col1) from `user` order by col1 asc limit :__upper_limit",
+			expField:    `[name:"col1" type:VARCHAR name:"sum(col2)" type:DECIMAL]`,
+			expRow:      `[[VARCHAR("a") DECIMAL(8)]]`,
+		},
+		{
+			sql:         `select col1, sum(col2) from (select col1, col2 from user limit 4) x group by col1`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col1)", "varchar|varchar|varbinary"), "a|x|a"),
+			expSandboxQ: "select col1, col2, weight_string(col1) from `user` order by col1 asc limit :__upper_limit",
+			expField:    `[name:"col1" type:VARCHAR name:"sum(col2)" type:DECIMAL]`,
+			expRow:      `[[VARCHAR("a") DECIMAL(0)]]`,
+		},
+		{
+			sql:         `select col1, sum(col2) from (select col1, col2 from user limit 4) x group by col1`,
+			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col1)", "varchar|varchar|varbinary"), "a|null|a"),
+			expSandboxQ: "select col1, col2, weight_string(col1) from `user` order by col1 asc limit :__upper_limit",
+			expField:    `[name:"col1" type:VARCHAR name:"sum(col2)" type:DECIMAL]`,
+			expRow:      `[[VARCHAR("a") NULL]]`,
+		},
+	}
+
+	for _, tc := range tcases {
+		t.Run(tc.sql, func(t *testing.T) {
+			for _, sbc := range conns {
+				sbc.SetResults([]*sqltypes.Result{tc.sandboxRes})
+				sbc.Queries = nil
+			}
+			qr, err := executorExec(executor, tc.sql, nil)
+			require.NoError(t, err)
+			assert.Equal(t, tc.expField, fmt.Sprintf("%v", qr.Fields))
+			assert.Equal(t, tc.expRow, fmt.Sprintf("%v", qr.Rows))
+			require.Len(t, conns[0].Queries, 1)
+			assert.Equal(t, tc.expSandboxQ, conns[0].Queries[0].Sql)
+		})
+	}
+}
diff --git a/go/vt/vtgate/executor_test.go b/go/vt/vtgate/executor_test.go
index b3e6e7c635..2f6278bded 100644
--- a/go/vt/vtgate/executor_test.go
+++ b/go/vt/vtgate/executor_test.go
@@ -746,7 +746,7 @@ func TestExecutorShow(t *testing.T) {
 	}
 	utils.MustMatch(t, wantqr, qr, query)
 
-	query = "show vitess_shards like 'TestSharded/%'"
+	query = "show vitess_shards like 'TestExecutor/%'"
 	qr, err = executor.Execute(ctx, "TestExecute", session, query, nil)
 	require.NoError(t, err)
 
@@ -755,13 +755,13 @@ func TestExecutorShow(t *testing.T) {
 	wantqr = &sqltypes.Result{
 		Fields: buildVarCharFields("Shards"),
 		Rows: [][]sqltypes.Value{
-			buildVarCharRow("TestSharded/-20"),
-			buildVarCharRow("TestSharded/e0-"),
+			buildVarCharRow("TestExecutor/-20"),
+			buildVarCharRow("TestExecutor/e0-"),
 		},
 	}
 	utils.MustMatch(t, wantqr, qr, query)
 
-	query = "show vitess_shards like 'TestShard%/%'"
+	query = "show vitess_shards like 'TestExec%/%'"
 	qr, err = executor.Execute(ctx, "TestExecute", session, query, nil)
 	require.NoError(t, err)
 
@@ -770,8 +770,8 @@ func TestExecutorShow(t *testing.T) {
 	wantqr = &sqltypes.Result{
 		Fields: buildVarCharFields("Shards"),
 		Rows: [][]sqltypes.Value{
-			buildVarCharRow("TestSharded/-20"),
-			buildVarCharRow("TestSharded/e0-"),
+			buildVarCharRow("TestExecutor/-20"),
+			buildVarCharRow("TestExecutor/e0-"),
 		},
 	}
 	utils.MustMatch(t, wantqr, qr, query)
@@ -844,7 +844,6 @@ func TestExecutorShow(t *testing.T) {
 			buildVarCharRow("TestExecutor", "keyspace_id", "numeric", "", ""),
 			buildVarCharRow("TestExecutor", "krcol_unique_vdx", "keyrange_lookuper_unique", "", ""),
 			buildVarCharRow("TestExecutor", "krcol_vdx", "keyrange_lookuper", "", ""),
-			buildVarCharRow("TestExecutor", "multicol_vdx", "multicol", "column_bytes=1,3,4; column_count=3; column_vindex=hash,binary,unicode_loose_xxhash", ""),
 			buildVarCharRow("TestExecutor", "music_user_map", "lookup_hash_unique", "from=music_id; table=music_user_map; to=user_id", "music"),
 			buildVarCharRow("TestExecutor", "name_lastname_keyspace_id_map", "lookup", "from=name,lastname; table=name_lastname_keyspace_id_map; to=keyspace_id", "user2"),
 			buildVarCharRow("TestExecutor", "name_user_map", "lookup_hash", "from=name; table=name_user_map; to=user_id", "user"),
@@ -852,6 +851,7 @@ func TestExecutorShow(t *testing.T) {
 			buildVarCharRow("TestExecutor", "t1_lkp_vdx", "consistent_lookup_unique", "from=unq_col; table=t1_lkp_idx; to=keyspace_id", "t1"),
 			buildVarCharRow("TestExecutor", "t2_lu_vdx", "lookup_hash_unique", "from=lu_col; table=TestUnsharded.lu_idx; to=keyspace_id", "t2_wo_lookup"),
 			buildVarCharRow("TestExecutor", "t2_wo_lu_vdx", "lookup_unique", "from=wo_lu_col; table=TestUnsharded.wo_lu_idx; to=keyspace_id; write_only=true", "t2_wo_lookup"),
+			buildVarCharRow("TestMultiCol", "multicol_vdx", "multicol", "column_bytes=1,3,4; column_count=3; column_vindex=hash,binary,unicode_loose_xxhash", ""),
 		},
 	}
 	utils.MustMatch(t, wantqr, qr, query)
@@ -967,7 +967,7 @@ func TestExecutorShow(t *testing.T) {
 	wantqr = &sqltypes.Result{
 		Fields: buildVarCharFields("Shards"),
 		Rows: [][]sqltypes.Value{
-			buildVarCharRow("TestSharded/-20"),
+			buildVarCharRow("TestMultiCol/-20"),
 			buildVarCharRow("TestXBadVSchema/e0-"),
 		},
 	}
diff --git a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
index a8ca9c018d..314d313f8e 100644
--- a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
@@ -107,7 +107,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(0)",
+    "Aggregates": "sum(0) AS count",
     "GroupBy": "1, 4, 3",
     "ResultColumns": 4,
     "Inputs": [
@@ -161,7 +161,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(0)",
+    "Aggregates": "sum(0) AS count",
     "GroupBy": "1",
     "Inputs": [
       {
@@ -218,7 +218,7 @@ Gen4 plan same as above
       {
         "OperatorType": "Aggregate",
         "Variant": "Ordered",
-        "Aggregates": "count(0)",
+        "Aggregates": "sum(0) AS count",
         "GroupBy": "1, 4, 3",
         "ResultColumns": 5,
         "Inputs": [
@@ -281,7 +281,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Scalar",
-    "Aggregates": "count(0)",
+    "Aggregates": "sum(0) AS count",
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -596,7 +596,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(1)",
+    "Aggregates": "sum(1) AS count",
     "GroupBy": "0",
     "Inputs": [
       {
@@ -676,7 +676,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(1)",
+    "Aggregates": "sum(1) AS count",
     "GroupBy": "0",
     "Inputs": [
       {
@@ -871,7 +871,7 @@ Gen4 plan same as above
       {
         "OperatorType": "Aggregate",
         "Variant": "Scalar",
-        "Aggregates": "count(0)",
+        "Aggregates": "sum(0) AS count",
         "Inputs": [
           {
             "OperatorType": "Route",
@@ -928,7 +928,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Scalar",
-    "Aggregates": "count(1)",
+    "Aggregates": "sum(1) AS count",
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -1048,7 +1048,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(1)",
+    "Aggregates": "sum(1) AS count",
     "GroupBy": "0",
     "Inputs": [
       {
@@ -1434,7 +1434,7 @@ Gen4 error: Can't group on 'count(*)'
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(2)",
+    "Aggregates": "sum(2) AS count",
     "GroupBy": "1, 0",
     "Inputs": [
       {
@@ -1487,7 +1487,7 @@ Gen4 error: Can't group on 'count(*)'
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(2)",
+    "Aggregates": "sum(2) AS count",
     "GroupBy": "1, 0",
     "Inputs": [
       {
@@ -1540,7 +1540,7 @@ Gen4 error: Can't group on 'count(*)'
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(2)",
+    "Aggregates": "sum(2) AS count",
     "GroupBy": "1, 0",
     "Inputs": [
       {
@@ -1647,7 +1647,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Scalar",
-    "Aggregates": "count(0)",
+    "Aggregates": "sum(0) AS count",
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -1694,7 +1694,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(4)",
+    "Aggregates": "sum(4) AS count",
     "GroupBy": "0, 1, 2",
     "Inputs": [
       {
@@ -1747,7 +1747,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(4)",
+    "Aggregates": "sum(4) AS count",
     "GroupBy": "0, 1, 2",
     "Inputs": [
       {
@@ -1800,7 +1800,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(4)",
+    "Aggregates": "sum(4) AS count",
     "GroupBy": "0, 1, 2, 3",
     "Inputs": [
       {
@@ -1853,7 +1853,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(4)",
+    "Aggregates": "sum(4) AS count",
     "GroupBy": "2, 1, 0, 3",
     "Inputs": [
       {
@@ -1906,7 +1906,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(3)",
+    "Aggregates": "sum(3) AS count",
     "GroupBy": "2, 1, 0",
     "Inputs": [
       {
@@ -1968,7 +1968,7 @@ Gen4 plan same as above
       {
         "OperatorType": "Aggregate",
         "Variant": "Ordered",
-        "Aggregates": "count(1)",
+        "Aggregates": "sum(1) AS count",
         "GroupBy": "0",
         "Inputs": [
           {
@@ -2091,7 +2091,7 @@ Gen4 plan same as above
       {
         "OperatorType": "Aggregate",
         "Variant": "Ordered",
-        "Aggregates": "count(1)",
+        "Aggregates": "sum(1) AS count",
         "GroupBy": "0",
         "Inputs": [
           {
@@ -2153,7 +2153,7 @@ Gen4 plan same as above
       {
         "OperatorType": "Aggregate",
         "Variant": "Ordered",
-        "Aggregates": "count(1)",
+        "Aggregates": "sum(1) AS count",
         "GroupBy": "0, 0",
         "Inputs": [
           {
@@ -2363,7 +2363,7 @@ Gen4 plan same as above
       {
         "OperatorType": "Aggregate",
         "Variant": "Scalar",
-        "Aggregates": "count(0)",
+        "Aggregates": "sum(0) AS count",
         "Inputs": [
           {
             "OperatorType": "Route",
@@ -2537,7 +2537,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(1)",
+    "Aggregates": "sum(1) AS count",
     "GroupBy": "0",
     "Inputs": [
       {
@@ -2607,7 +2607,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Scalar",
-    "Aggregates": "count(0)",
+    "Aggregates": "sum(0) AS count",
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -2673,7 +2673,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(1)",
+    "Aggregates": "sum(1) AS count",
     "GroupBy": "0",
     "Inputs": [
       {
@@ -2726,7 +2726,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(1)",
+    "Aggregates": "sum(1) AS count",
     "GroupBy": "0",
     "Inputs": [
       {
@@ -3740,7 +3740,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count_distinct(1) AS count(distinct val1), count(2)",
+    "Aggregates": "count_distinct(1) AS count(distinct val1), sum(2) AS count",
     "GroupBy": "0",
     "Inputs": [
       {
@@ -3793,7 +3793,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(1)",
+    "Aggregates": "sum(1) AS count",
     "GroupBy": "0",
     "Inputs": [
       {
@@ -4635,7 +4635,6 @@ Gen4 error: aggregate functions take a single argument 'count(distinct user_id,
   }
 }
 
-
 # scatter aggregate symtab lookup error
 "select id, b as id, count(*) from user order by id"
 "ambiguous symbol reference: id"
@@ -4672,7 +4671,7 @@ Gen4 error: aggregate functions take a single argument 'count(distinct user_id,
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Scalar",
-    "Aggregates": "count(1)",
+    "Aggregates": "sum(1) AS count",
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -4768,3 +4767,145 @@ Gen4 error: aggregate functions take a single argument 'count(distinct user_id,
     ]
   }
 }
+
+# count on column from LIMIT
+"select count(city) from (select phone, id, city from user where id > 12 limit 10) as x"
+"unsupported: cross-shard query with aggregates"
+{
+  "QueryType": "SELECT",
+  "Original": "select count(city) from (select phone, id, city from user where id \u003e 12 limit 10) as x",
+  "Instructions": {
+    "OperatorType": "Aggregate",
+    "Variant": "Scalar",
+    "Aggregates": "count(0) AS count(city)",
+    "Inputs": [
+      {
+        "OperatorType": "Projection",
+        "Expressions": [
+          "[COLUMN 2] as count(city)"
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "Limit",
+            "Count": "INT64(10)",
+            "Inputs": [
+              {
+                "OperatorType": "Route",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FieldQuery": "select phone, id, city from `user` where 1 != 1",
+                "Query": "select phone, id, city from `user` where id \u003e 12 limit :__upper_limit",
+                "Table": "`user`"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  }
+}
+
+# count(*) on column from LIMIT
+"select count(*) from (select phone, id, city from user where id > 12 limit 10) as x"
+"unsupported: cross-shard query with aggregates"
+{
+  "QueryType": "SELECT",
+  "Original": "select count(*) from (select phone, id, city from user where id \u003e 12 limit 10) as x",
+  "Instructions": {
+    "OperatorType": "Aggregate",
+    "Variant": "Scalar",
+    "Aggregates": "count_star(0) AS count(*)",
+    "Inputs": [
+      {
+        "OperatorType": "Projection",
+        "Expressions": [
+          "[COLUMN 0] as count(*)"
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "Limit",
+            "Count": "INT64(10)",
+            "Inputs": [
+              {
+                "OperatorType": "Route",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FieldQuery": "select phone, id, city from `user` where 1 != 1",
+                "Query": "select phone, id, city from `user` where id \u003e 12 limit :__upper_limit",
+                "Table": "`user`"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  }
+}
+
+# count non-null columns incoming from outer joins should work well
+"select count(col) from (select user_extra.col as col from user left join user_extra on user.id = user_extra.id limit 10) as x"
+{
+  "QueryType": "SELECT",
+  "Original": "select count(col) from (select user_extra.col as col from user left join user_extra on user.id = user_extra.id limit 10) as x",
+  "Instructions": {
+    "OperatorType": "Aggregate",
+    "Variant": "Scalar",
+    "Aggregates": "count(0) AS count(col)",
+    "Inputs": [
+      {
+        "OperatorType": "Projection",
+        "Expressions": [
+          "[COLUMN 0] as count(col)"
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "Limit",
+            "Count": "INT64(10)",
+            "Inputs": [
+              {
+                "OperatorType": "Join",
+                "Variant": "LeftJoin",
+                "JoinColumnIndexes": "R:0",
+                "JoinVars": {
+                  "user_id": 0
+                },
+                "TableName": "`user`_user_extra",
+                "Inputs": [
+                  {
+                    "OperatorType": "Route",
+                    "Variant": "Scatter",
+                    "Keyspace": {
+                      "Name": "user",
+                      "Sharded": true
+                    },
+                    "FieldQuery": "select `user`.id from `user` where 1 != 1",
+                    "Query": "select `user`.id from `user`",
+                    "Table": "`user`"
+                  },
+                  {
+                    "OperatorType": "Route",
+                    "Variant": "Scatter",
+                    "Keyspace": {
+                      "Name": "user",
+                      "Sharded": true
+                    },
+                    "FieldQuery": "select user_extra.col as col from user_extra where 1 != 1",
+                    "Query": "select user_extra.col as col from user_extra where user_extra.id = :user_id",
+                    "Table": "user_extra"
+                  }
+                ]
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  }
+}
+Gen4 plan same as above
diff --git a/go/vt/vtgate/planbuilder/testdata/from_cases.txt b/go/vt/vtgate/planbuilder/testdata/from_cases.txt
index 72510897cf..9a2e8dfbb5 100644
--- a/go/vt/vtgate/planbuilder/testdata/from_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/from_cases.txt
@@ -3698,7 +3698,7 @@ Gen4 plan same as above
       {
         "OperatorType": "Aggregate",
         "Variant": "Scalar",
-        "Aggregates": "count(0)",
+        "Aggregates": "sum(0) AS count",
         "Inputs": [
           {
             "OperatorType": "Route",
@@ -4623,7 +4623,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(1)",
+    "Aggregates": "sum(1) AS count",
     "GroupBy": "0",
     "Inputs": [
       {
diff --git a/go/vt/vtgate/planbuilder/testdata/memory_sort_cases.txt b/go/vt/vtgate/planbuilder/testdata/memory_sort_cases.txt
index d117545164..b9a8e36b96 100644
--- a/go/vt/vtgate/planbuilder/testdata/memory_sort_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/memory_sort_cases.txt
@@ -13,7 +13,7 @@
       {
         "OperatorType": "Aggregate",
         "Variant": "Ordered",
-        "Aggregates": "count(2)",
+        "Aggregates": "sum(2) AS count",
         "GroupBy": "0",
         "ResultColumns": 4,
         "Inputs": [
@@ -74,7 +74,7 @@
       {
         "OperatorType": "Aggregate",
         "Variant": "Ordered",
-        "Aggregates": "count(2)",
+        "Aggregates": "sum(2) AS count",
         "GroupBy": "0",
         "Inputs": [
           {
@@ -141,7 +141,7 @@
       {
         "OperatorType": "Aggregate",
         "Variant": "Ordered",
-        "Aggregates": "count(2)",
+        "Aggregates": "sum(2) AS count",
         "GroupBy": "0",
         "ResultColumns": 5,
         "Inputs": [
@@ -213,7 +213,7 @@
           {
             "OperatorType": "Aggregate",
             "Variant": "Ordered",
-            "Aggregates": "count(2)",
+            "Aggregates": "sum(2) AS count",
             "GroupBy": "0",
             "Inputs": [
               {
@@ -289,7 +289,7 @@
       {
         "OperatorType": "Aggregate",
         "Variant": "Ordered",
-        "Aggregates": "count(2)",
+        "Aggregates": "sum(2) AS count",
         "GroupBy": "0",
         "ResultColumns": 4,
         "Inputs": [
@@ -359,7 +359,7 @@
       {
         "OperatorType": "Aggregate",
         "Variant": "Ordered",
-        "Aggregates": "count(1)",
+        "Aggregates": "sum(1) AS count",
         "GroupBy": "2",
         "ResultColumns": 3,
         "Inputs": [
diff --git a/go/vt/vtgate/planbuilder/testdata/postprocess_cases.txt b/go/vt/vtgate/planbuilder/testdata/postprocess_cases.txt
index af723e5177..eb1b28db5f 100644
--- a/go/vt/vtgate/planbuilder/testdata/postprocess_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/postprocess_cases.txt
@@ -1837,7 +1837,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Scalar",
-    "Aggregates": "count(0)",
+    "Aggregates": "sum(0) AS count",
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -1890,7 +1890,7 @@ Gen4 plan same as above
       {
         "OperatorType": "Aggregate",
         "Variant": "Scalar",
-        "Aggregates": "count(0)",
+        "Aggregates": "sum(0) AS count",
         "ResultColumns": 3,
         "Inputs": [
           {
@@ -1942,7 +1942,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "count(0)",
+    "Aggregates": "sum(0) AS count",
     "GroupBy": "1",
     "Inputs": [
       {
@@ -2000,7 +2000,7 @@ Gen4 plan same as above
       {
         "OperatorType": "Aggregate",
         "Variant": "Ordered",
-        "Aggregates": "count(0)",
+        "Aggregates": "sum(0) AS count",
         "GroupBy": "1",
         "Inputs": [
           {
@@ -2123,7 +2123,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Ordered",
-    "Aggregates": "sum(1) AS count(*), random(2)",
+    "Aggregates": "sum(1) AS count(*), random(2) AS c1",
     "GroupBy": "0",
     "ResultColumns": 2,
     "Inputs": [
diff --git a/go/vt/vtgate/planbuilder/testdata/select_cases.txt b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
index 5d5c4a0ccc..cf760b9828 100644
--- a/go/vt/vtgate/planbuilder/testdata/select_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
@@ -83,7 +83,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Scalar",
-    "Aggregates": "count(0)",
+    "Aggregates": "sum(0) AS count",
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -178,7 +178,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Scalar",
-    "Aggregates": "count(0)",
+    "Aggregates": "sum(0) AS count",
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -227,7 +227,7 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "Aggregate",
     "Variant": "Scalar",
-    "Aggregates": "count(0)",
+    "Aggregates": "sum(0) AS count",
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -1758,7 +1758,7 @@ Gen4 plan same as above
       {
         "OperatorType": "Aggregate",
         "Variant": "Scalar",
-        "Aggregates": "count(0)",
+        "Aggregates": "sum(0) AS count",
         "Inputs": [
           {
             "OperatorType": "Route",
@@ -1894,7 +1894,7 @@ Gen4 plan same as above
       {
         "OperatorType": "Aggregate",
         "Variant": "Scalar",
-        "Aggregates": "count(0)",
+        "Aggregates": "sum(0) AS count",
         "Inputs": [
           {
             "OperatorType": "Route",
diff --git a/go/vt/vtgate/planbuilder/testdata/union_cases.txt b/go/vt/vtgate/planbuilder/testdata/union_cases.txt
index 60c1531f1e..2e75846afa 100644
--- a/go/vt/vtgate/planbuilder/testdata/union_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/union_cases.txt
@@ -1270,7 +1270,7 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
           {
             "OperatorType": "Aggregate",
             "Variant": "Scalar",
-            "Aggregates": "count(0)",
+            "Aggregates": "sum(0) AS count",
             "Inputs": [
               {
                 "OperatorType": "Route",
@@ -1288,7 +1288,7 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
           {
             "OperatorType": "Aggregate",
             "Variant": "Scalar",
-            "Aggregates": "count(0)",
+            "Aggregates": "sum(0) AS count",
             "Inputs": [
               {
                 "OperatorType": "Route",
diff --git a/go/vt/vtgate/sandbox_test.go b/go/vt/vtgate/sandbox_test.go
index 01165eaee6..4f7a30dbcc 100644
--- a/go/vt/vtgate/sandbox_test.go
+++ b/go/vt/vtgate/sandbox_test.go
@@ -40,7 +40,7 @@ import (
 // sandbox_test.go provides a sandbox for unit testing VTGate.
 
 const (
-	KsTestSharded             = "TestSharded"
+	KsTestSharded             = "TestExecutor"
 	KsTestUnsharded           = "TestUnsharded"
 	KsTestUnshardedServedFrom = "TestUnshardedServedFrom"
 	KsTestBadVSchema          = "TestXBadVSchema"
diff --git a/go/vt/vtgate/vtgate_test.go b/go/vt/vtgate/vtgate_test.go
index c2a53b762e..f68b7fda71 100644
--- a/go/vt/vtgate/vtgate_test.go
+++ b/go/vt/vtgate/vtgate_test.go
@@ -52,7 +52,7 @@ var executeOptions = &querypb.ExecuteOptions{
 var primarySession *vtgatepb.Session
 
 func init() {
-	getSandbox(KsTestUnsharded).VSchema = `
+	createSandbox(KsTestUnsharded).VSchema = `
 {
 	"sharded": false,
 	"tables": {
@@ -60,7 +60,7 @@ func init() {
 	}
 }
 `
-	getSandbox(KsTestBadVSchema).VSchema = `
+	createSandbox(KsTestBadVSchema).VSchema = `
 {
 	"sharded": true,
 	"tables": {
diff --git a/go/vt/wrangler/vdiff_test.go b/go/vt/wrangler/vdiff_test.go
index 4f1574cc85..dffe3a4abb 100644
--- a/go/vt/wrangler/vdiff_test.go
+++ b/go/vt/wrangler/vdiff_test.go
@@ -395,7 +395,7 @@ func TestVDiffPlanSuccess(t *testing.T) {
 			selectPks:        []int{0},
 			sourcePrimitive: &engine.OrderedAggregate{
 				Aggregates: []*engine.AggregateParams{{
-					Opcode: engine.AggregateCount,
+					Opcode: engine.AggregateSum,
 					Col:    2,
 				}, {
 					Opcode: engine.AggregateSum,