Repository: vitessio/vitess
Pull request #10373 (2022-05-27 09:04:18)

vt/wrangler: fix deleting primary tablet if `Shard` does not exist
This PR makes sure that `wrangler.DeleteTablet()` no longer returns an error when deleting a Primary if a `Shard` no longer exists. A `Shard` might not exist, when for example someone explicitly deletes it before calling `wrangler.DeleteTablet()`. By making this change, the `wrangler.DeleteTablet()` function is also idempotent, and safe to use in reconciling loops (such as in Kubernetes Operators).


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.
fatih: @rohit-nayak-ps I see. It looks like `isPrimaryTablet`is also calling `GetShard`, so it's worth adding the same modification there too. I'm going to push another fix now. 
fatih: It passes locally:

```
$ go test
...
PASS
ok      vitess.io/vitess/go/vt/wrangler 5.003s
```

But I'm not sure why it's failing in the CI.


Patch:
diff --git a/go/vt/vtadmin/cluster/cluster.go b/go/vt/vtadmin/cluster/cluster.go
index 17fa0438fe..b7087677b0 100644
--- a/go/vt/vtadmin/cluster/cluster.go
+++ b/go/vt/vtadmin/cluster/cluster.go
@@ -211,14 +211,20 @@ func New(ctx context.Context, cfg Config) (*Cluster, error) {
 // to avoid data races in tests (the latter of these is caused by the cache
 // goroutines).
 //
-// Sub-components of the cluster are `Close`-d concurrently.
+// Sub-components of the cluster are `Close`-d concurrently, caches first, then
+// proxy connections.
 func (c *Cluster) Close() error {
 	var (
 		wg  sync.WaitGroup
 		rec concurrency.AllErrorRecorder
 	)
 
-	for _, closer := range []io.Closer{c.DB, c.Vtctld, c.schemaCache} {
+	// First, close any caches, which may have connections to DB or Vtctld
+	// (N.B. (andrew) when we have multiple caches, we can close them
+	// concurrently, like we do with the proxies).
+	rec.RecordError(c.schemaCache.Close())
+
+	for _, closer := range []io.Closer{c.DB, c.Vtctld} {
 		wg.Add(1)
 		go func(closer io.Closer) {
 			defer wg.Done()
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/tablet.go b/go/vt/wrangler/tablet.go
index 94454874be..bc69fb5de3 100644
--- a/go/vt/wrangler/tablet.go
+++ b/go/vt/wrangler/tablet.go
@@ -46,7 +46,7 @@ func (wr *Wrangler) DeleteTablet(ctx context.Context, tabletAlias *topodatapb.Ta
 	}
 
 	wasPrimary, err := wr.isPrimaryTablet(ctx, ti)
-	if err != nil {
+	if err != nil && !topo.IsErrType(err, topo.NoNode) {
 		return err
 	}
 
@@ -65,7 +65,7 @@ func (wr *Wrangler) DeleteTablet(ctx context.Context, tabletAlias *topodatapb.Ta
 		defer unlock(&err)
 
 		// update the shard record's primary
-		if _, err := wr.ts.UpdateShardFields(ctx, ti.Keyspace, ti.Shard, func(si *topo.ShardInfo) error {
+		_, err := wr.ts.UpdateShardFields(ctx, ti.Keyspace, ti.Shard, func(si *topo.ShardInfo) error {
 			if !topoproto.TabletAliasEqual(si.PrimaryAlias, tabletAlias) {
 				wr.Logger().Warningf("Deleting primary %v from shard %v/%v but primary in Shard object was %v", topoproto.TabletAliasString(tabletAlias), ti.Keyspace, ti.Shard, topoproto.TabletAliasString(si.PrimaryAlias))
 				return topo.NewError(topo.NoUpdateNeeded, si.Keyspace()+"/"+si.ShardName())
@@ -73,7 +73,8 @@ func (wr *Wrangler) DeleteTablet(ctx context.Context, tabletAlias *topodatapb.Ta
 			si.PrimaryAlias = nil
 			si.SetPrimaryTermStartTime(time.Now())
 			return nil
-		}); err != nil {
+		})
+		if err != nil && !topo.IsErrType(err, topo.NoNode) {
 			return err
 		}
 	}
@@ -107,7 +108,6 @@ func (wr *Wrangler) ChangeTabletType(ctx context.Context, tabletAlias *topodatap
 	expectedTablet := proto.Clone(ti.Tablet).(*topodatapb.Tablet)
 	expectedTablet.Type = tabletType
 	semiSync, err := wr.shouldSendSemiSyncAck(ctx, expectedTablet)
-
 	if err != nil {
 		return err
 	}
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/tablet_test.go b/go/vt/wrangler/tablet_test.go
index 881cff7b1c..b41f261170 100644
--- a/go/vt/wrangler/tablet_test.go
+++ b/go/vt/wrangler/tablet_test.go
@@ -17,11 +17,10 @@ limitations under the License.
 package wrangler
 
 import (
+	"context"
 	"strings"
 	"testing"
 
-	"context"
-
 	"vitess.io/vitess/go/vt/logutil"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	"vitess.io/vitess/go/vt/topo"
@@ -179,3 +178,47 @@ func TestDeleteTabletFalsePrimary(t *testing.T) {
 		t.Fatalf("DeleteTablet failed: %v", err)
 	}
 }
+
+// TestDeleteTabletShardNonExisting tests that you can delete a true primary
+// tablet if a shard does not exists anymore.
+func TestDeleteTabletShardNonExisting(t *testing.T) {
+	cell := "cell1"
+	ts := memorytopo.NewServer(cell)
+	wr := New(logutil.NewConsoleLogger(), ts, nil)
+
+	tablet := &topodatapb.Tablet{
+		Alias: &topodatapb.TabletAlias{
+			Cell: cell,
+			Uid:  1,
+		},
+		Keyspace: "test",
+		Shard:    "0",
+		Type:     topodatapb.TabletType_PRIMARY,
+	}
+
+	if err := wr.TopoServer().InitTablet(context.Background(), tablet, false /*allowPrimaryOverride*/, true /*createShardAndKeyspace*/, false /*allowUpdate*/); err != nil {
+		t.Fatalf("InitTablet failed: %v", err)
+	}
+	if _, err := ts.GetTablet(context.Background(), tablet.Alias); err != nil {
+		t.Fatalf("GetTablet failed: %v", err)
+	}
+
+	// set PrimaryAlias and PrimaryTermStartTime on shard to match chosen primary tablet
+	if _, err := ts.UpdateShardFields(context.Background(), "test", "0", func(si *topo.ShardInfo) error {
+		si.PrimaryAlias = tablet.Alias
+		si.PrimaryTermStartTime = tablet.PrimaryTermStartTime
+		return nil
+	}); err != nil {
+		t.Fatalf("UpdateShardFields failed: %v", err)
+	}
+
+	// trigger a shard deletion
+	if err := ts.DeleteShard(context.Background(), "test", "0"); err != nil {
+		t.Fatalf("DeleteShard failed: %v", err)
+	}
+
+	// DeleteTablet should not fail if a shard no longer exist
+	if err := wr.DeleteTablet(context.Background(), tablet.Alias, true); err != nil {
+		t.Fatalf("DeleteTablet failed: %v", err)
+	}
+}
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,