Repository: vitessio/vitess
Pull request #9765 (2022-02-23 14:36:50)

Filter candidates which cannot make forward progress after ERS
## Description
<!-- A few sentences describing the overall goals of the pull request's commits. -->
<!-- If this is a bug fix and you think the fix should be backported, please write so. -->
After the changes in #9659, we started allowing multiple failures in ERS as long as we are able to guarantee safety.
This leads to an additional concern of a tablet being promoted which cannot make forward progress after being elected.

Consider the following case - 
We have 5 tablets (A, B, C, D and E) of which A, B and C are eligible to be primary tablets with C being the current primary. Let the durability policies be configured such that A requires 3 semi-sync acks to accept transactions  and both B and C require only 2. 

Consider the case when C and D both become unreachable. In this case we have more than 1 failures and pre #9659 we would have failed ERS. Post #9659, we will try to recover since having A, B and E reachable is enough to know that C won't be able to accept any writes, so it is safe to proceed.

In this case, if we promote tablet A, we will have promoted a tablet which cannot make forward progress (there are only 2 semi-sync acking tablets B and E available) and will hang on all the writes. Instead we must promote B.

This PR fixes this issue by creating a function analogous to Revocation, called Establish which checks whether a tablet can make forward progress given the list of reachable tablets.

This function is then used in ERS to filter out all the tablets which should not be promoted before we select a new tablet.
As part of this PR, other constraints checks have also been moved into a common function to filter primary candidates.  

## Related Issue(s)
<!-- List related issues and pull requests. If this PR fixes an issue, please add it using Fixes #????  -->
- #8975 

## Checklist
- [x] Should this PR be backported? *No*
- [x] Tests were added or are not required
- [x] Documentation was added or is not required

## Deployment Notes
<!-- Notes regarding deployment of the contained body of work. These should note any db migrations, etc. -->


Patch:
diff --git a/go/vt/vtctl/reparentutil/durability_funcs.go b/go/vt/vtctl/reparentutil/durability_funcs.go
index 2d4ead1f6f..b9b76fe0b4 100644
--- a/go/vt/vtctl/reparentutil/durability_funcs.go
+++ b/go/vt/vtctl/reparentutil/durability_funcs.go
@@ -72,3 +72,22 @@ func haveRevoked(tabletsReached []*topodatapb.Tablet, allTablets []*topodatapb.T
 	}
 	return true
 }
+
+// canEstablishForTablet checks whether we have reached enough tablets to say that the given primary eligible tablet will be able to accept new writes
+func canEstablishForTablet(primaryEligible *topodatapb.Tablet, tabletsReached []*topodatapb.Tablet) bool {
+	// if we have not reached the primaryEligible tablet, then it cannot be considered eligible to accept writes
+	// since it might have been stopped
+	if !topoproto.IsTabletInList(primaryEligible, tabletsReached) {
+		return false
+	}
+
+	// semiSyncAckersReached is the list of reachable tablets capable of sending semi sync Acks for the given primaryEligible tablet
+	semiSyncAckersReached := SemiSyncAckersForPrimary(primaryEligible, tabletsReached)
+
+	// numOfSemiSyncAcksRequired is the number of semi sync Acks that the primaryEligible tablet requires
+	numOfSemiSyncAcksRequired := SemiSyncAckers(primaryEligible)
+
+	// if we have reached enough semi-sync Acking tablets such that the primaryEligible can accept a write
+	// we can safely promote this tablet
+	return len(semiSyncAckersReached) >= numOfSemiSyncAcksRequired
+}
diff --git a/go/vt/vtctl/reparentutil/emergency_reparenter.go b/go/vt/vtctl/reparentutil/emergency_reparenter.go
index 2b20444c99..ff97fe2f22 100644
--- a/go/vt/vtctl/reparentutil/emergency_reparenter.go
+++ b/go/vt/vtctl/reparentutil/emergency_reparenter.go
@@ -139,11 +139,10 @@ func (erp *EmergencyReparenter) reparentShardLocked(ctx context.Context, ev *eve
 	ersCounter.Add(1)
 
 	var (
+		stoppedReplicationSnapshot *replicationSnapshot
 		shardInfo                  *topo.ShardInfo
 		prevPrimary                *topodatapb.Tablet
 		tabletMap                  map[string]*topo.TabletInfo
-		statusMap                  map[string]*replicationdatapb.StopReplicationStatus
-		primaryStatusMap           map[string]*replicationdatapb.PrimaryStatus
 		validCandidates            map[string]mysql.Position
 		intermediateSource         *topodatapb.Tablet
 		validCandidateTablets      []*topodatapb.Tablet
@@ -177,7 +176,7 @@ func (erp *EmergencyReparenter) reparentShardLocked(ctx context.Context, ev *eve
 	}
 
 	// Stop replication on all the tablets and build their status map
-	statusMap, primaryStatusMap, err = StopReplicationAndBuildStatusMaps(ctx, erp.tmc, ev, tabletMap, opts.WaitReplicasTimeout, opts.IgnoreReplicas, opts.NewPrimaryAlias, erp.logger)
+	stoppedReplicationSnapshot, err = stopReplicationAndBuildStatusMaps(ctx, erp.tmc, ev, tabletMap, opts.WaitReplicasTimeout, opts.IgnoreReplicas, opts.NewPrimaryAlias, erp.logger)
 	if err != nil {
 		return vterrors.Wrapf(err, "failed to stop replication and build status maps: %v", err)
 	}
@@ -189,7 +188,7 @@ func (erp *EmergencyReparenter) reparentShardLocked(ctx context.Context, ev *eve
 
 	// find the valid candidates for becoming the primary
 	// this is where we check for errant GTIDs and remove the tablets that have them from consideration
-	validCandidates, err = FindValidEmergencyReparentCandidates(statusMap, primaryStatusMap)
+	validCandidates, err = FindValidEmergencyReparentCandidates(stoppedReplicationSnapshot.statusMap, stoppedReplicationSnapshot.primaryStatusMap)
 	if err != nil {
 		return err
 	}
@@ -202,7 +201,7 @@ func (erp *EmergencyReparenter) reparentShardLocked(ctx context.Context, ev *eve
 	}
 
 	// Wait for all candidates to apply relay logs
-	if err = erp.waitForAllRelayLogsToApply(ctx, validCandidates, tabletMap, statusMap, opts.WaitReplicasTimeout); err != nil {
+	if err = erp.waitForAllRelayLogsToApply(ctx, validCandidates, tabletMap, stoppedReplicationSnapshot.statusMap, opts.WaitReplicasTimeout); err != nil {
 		return err
 	}
 
@@ -211,13 +210,25 @@ func (erp *EmergencyReparenter) reparentShardLocked(ctx context.Context, ev *eve
 	// In case the user has specified a tablet specifically, then it is selected, as long as it is the most advanced.
 	// Here we also check for split brain scenarios and check that the selected replica must be more advanced than all the other valid candidates.
 	// We fail in case there is a split brain detected.
+	// The validCandidateTablets list is sorted by the replication positions with ties broken by promotion rules.
 	intermediateSource, validCandidateTablets, err = erp.findMostAdvanced(validCandidates, tabletMap, opts)
 	if err != nil {
 		return err
 	}
 	erp.logger.Infof("intermediate source selected - %v", intermediateSource.Alias)
 
-	// check whether the intermediate source candidate selected is ideal or if it can be improved later
+	// After finding the intermediate source, we want to filter the valid candidate list by the following criteria -
+	// 1. Only keep the tablets which can make progress after being promoted (have sufficient reachable semi-sync ackers)
+	// 2. Remove the tablets with the Must_not promote rule
+	// 3. Remove cross-cell tablets if PreventCrossCellPromotion is specified
+	// Our final primary candidate MUST belong to this list of valid candidates
+	validCandidateTablets, err = erp.filterValidCandidates(validCandidateTablets, stoppedReplicationSnapshot.reachableTablets, prevPrimary, opts)
+	if err != nil {
+		return err
+	}
+
+	// Check whether the intermediate source candidate selected is ideal or if it can be improved later.
+	// If the intermediateSource is ideal, then we can be certain that it is part of the valid candidates list.
 	isIdeal, err = erp.intermediateSourceIsIdeal(intermediateSource, prevPrimary, validCandidateTablets, tabletMap, opts)
 	if err != nil {
 		return err
@@ -234,8 +245,9 @@ func (erp *EmergencyReparenter) reparentShardLocked(ctx context.Context, ev *eve
 	if !isIdeal {
 		// we now reparent all the tablets to start replicating from the intermediate source
 		// we do not promote the tablet or change the shard record. We only change the replication for all the other tablets
-		// it also returns the list of the tablets that started replication successfully including itself. These are the candidates that we can use to find a replacement
-		validReplacementCandidates, err = erp.promoteIntermediateSource(ctx, ev, intermediateSource, tabletMap, statusMap, opts)
+		// it also returns the list of the tablets that started replication successfully including itself part of the validCandidateTablets list.
+		// These are the candidates that we can use to find a replacement.
+		validReplacementCandidates, err = erp.promoteIntermediateSource(ctx, ev, intermediateSource, tabletMap, stoppedReplicationSnapshot.statusMap, validCandidateTablets, opts)
 		if err != nil {
 			return err
 		}
@@ -258,27 +270,16 @@ func (erp *EmergencyReparenter) reparentShardLocked(ctx context.Context, ev *eve
 		}
 	}
 
-	// now we check if all the constraints are satisfied. If they are not, then we should exit
-	constraintFailure := erp.checkIfConstraintsSatisfied(newPrimary, prevPrimary, opts)
-	if constraintFailure != nil {
-		erp.logger.Errorf("have to override promotion because of constraint failure - %v", constraintFailure)
-		// we want to send both the errors to the user, constraint failure and also any error encountered in undoing the promotion
-		defer func() {
-			if err != nil {
-				err = vterrors.Errorf(vtrpc.Code_ABORTED, "error in undoing promotion - %v, constraint failure - %v", err, constraintFailure)
-			} else {
-				err = constraintFailure
-			}
-		}()
-		// we now try to undo are changes. We can do so by promoting the previous primary instead of the new one we selected
-		if prevPrimary == nil {
-			return vterrors.Errorf(vtrpc.Code_ABORTED, "could not undo promotion, since shard record has no primary information")
-		}
-		newPrimary = prevPrimary
-	}
+	// The new primary which will be promoted will always belong to the validCandidateTablets list because -
+	// 	1. 	if the intermediate source is ideal - then we know the intermediate source was in the validCandidateTablets list
+	// 		since we used that list
+	//	2. 	if the intermediate source isn't ideal - we take the intersection of the validCandidateTablets list and the one we
+	//		were able to reach during the promotion of intermediate source, as possible candidates. So the final candidate (even if
+	//		it is the intermediate source itself) will belong to the list
+	// Since the new primary tablet belongs to the validCandidateTablets list, we no longer need any additional constraint checks
 
 	// Final step is to promote our primary candidate
-	err = erp.promoteNewPrimary(ctx, ev, newPrimary, opts, tabletMap, statusMap)
+	err = erp.promoteNewPrimary(ctx, ev, newPrimary, opts, tabletMap, stoppedReplicationSnapshot.statusMap)
 	if err != nil {
 		return err
 	}
@@ -303,7 +304,7 @@ func (erp *EmergencyReparenter) waitForAllRelayLogsToApply(
 	waiterCount := 0
 
 	for candidate := range validCandidates {
-		// When we called StopReplicationAndBuildStatusMaps, we got back two
+		// When we called stopReplicationAndBuildStatusMaps, we got back two
 		// maps: (1) the StopReplicationStatus of any replicas that actually
 		// stopped replication; and (2) the PrimaryStatus of anything that
 		// returned ErrNotReplica, which is a tablet that is either the current
@@ -417,17 +418,28 @@ func (erp *EmergencyReparenter) promoteIntermediateSource(
 	source *topodatapb.Tablet,
 	tabletMap map[string]*topo.TabletInfo,
 	statusMap map[string]*replicationdatapb.StopReplicationStatus,
+	validCandidateTablets []*topodatapb.Tablet,
 	opts EmergencyReparentOptions,
 ) ([]*topodatapb.Tablet, error) {
 	// we reparent all the other tablets to start replication from our new source
 	// we wait for all the replicas so that we can choose a better candidate from the ones that started replication later
-	validCandidatesForImprovement, err := erp.reparentReplicas(ctx, ev, source, tabletMap, statusMap, opts, true /* waitForAllReplicas */, false /* populateReparentJournal */)
+	reachableTablets, err := erp.reparentReplicas(ctx, ev, source, tabletMap, statusMap, opts, true /* waitForAllReplicas */, false /* populateReparentJournal */)
 	if err != nil {
 		return nil, err
 	}
 
 	// also include the current tablet for being considered as part of valid candidates for ERS promotion
-	validCandidatesForImprovement = append(validCandidatesForImprovement, source)
+	reachableTablets = append(reachableTablets, source)
+
+	// The only valid candidates for improvement are the ones which are reachable and part of the valid candidate list.
+	// Here we need to be careful not to mess up the ordering of tablets in validCandidateTablets, since the list is sorted by the
+	// replication positions.
+	var validCandidatesForImprovement []*topodatapb.Tablet
+	for _, tablet := range validCandidateTablets {
+		if topoproto.IsTabletInList(tablet, reachableTablets) {
+			validCandidatesForImprovement = append(validCandidatesForImprovement, tablet)
+		}
+	}
 	return validCandidatesForImprovement, nil
 }
 
@@ -619,6 +631,10 @@ func (erp *EmergencyReparenter) identifyPrimaryCandidate(
 		}
 	}()
 
+	if len(validCandidates) == 0 {
+		return nil, vterrors.Errorf(vtrpc.Code_FAILED_PRECONDITION, "no valid candidates for emergency reparent")
+	}
+
 	if opts.NewPrimaryAlias != nil {
 		// explicit request to promote a specific tablet
 		requestedPrimaryAlias := topoproto.TabletAliasString(opts.NewPrimaryAlias)
@@ -689,19 +705,8 @@ func (erp *EmergencyReparenter) identifyPrimaryCandidate(
 		}
 	}
 
-	// return the one that we have if nothing is found
-	return intermediateSource, nil
-}
-
-// checkIfConstraintsSatisfied is used to check whether the constraints for ERS are satisfied or not.
-func (erp *EmergencyReparenter) checkIfConstraintsSatisfied(newPrimary, prevPrimary *topodatapb.Tablet, opts EmergencyReparentOptions) error {
-	if opts.PreventCrossCellPromotion && prevPrimary != nil && newPrimary.Alias.Cell != prevPrimary.Alias.Cell {
-		return vterrors.Errorf(vtrpc.Code_ABORTED, "elected primary does not satisfy geographic constraint - %s", topoproto.TabletAliasString(newPrimary.Alias))
-	}
-	if PromotionRule(newPrimary) == promotionrule.MustNot {
-		return vterrors.Errorf(vtrpc.Code_ABORTED, "elected primary does not satisfy promotion rule constraint - %s", topoproto.TabletAliasString(newPrimary.Alias))
-	}
-	return nil
+	// return any valid tablet
+	return findCandidateAnyCell(intermediateSource, validCandidates), nil
 }
 
 func (erp *EmergencyReparenter) promoteNewPrimary(
@@ -733,3 +738,37 @@ func (erp *EmergencyReparenter) promoteNewPrimary(
 	}
 	return nil
 }
+
+// filterValidCandidates filters valid tablets, keeping only the ones which can successfully be promoted without any constraint failures and can make forward progress on being promoted
+func (erp *EmergencyReparenter) filterValidCandidates(validTablets []*topodatapb.Tablet, tabletsReachable []*topodatapb.Tablet, prevPrimary *topodatapb.Tablet, opts EmergencyReparentOptions) ([]*topodatapb.Tablet, error) {
+	var restrictedValidTablets []*topodatapb.Tablet
+	for _, tablet := range validTablets {
+		tabletAliasStr := topoproto.TabletAliasString(tablet.Alias)
+		// Remove tablets which have MustNot promote rule since they must never be promoted
+		if PromotionRule(tablet) == promotionrule.MustNot {
+			erp.logger.Infof("Removing %s from list of valid candidates for promotion because it has the Must Not promote rule", tabletAliasStr)
+			if opts.NewPrimaryAlias != nil && topoproto.TabletAliasEqual(opts.NewPrimaryAlias, tablet.Alias) {
+				return nil, vterrors.Errorf(vtrpc.Code_ABORTED, "proposed primary %s has a must not promotion rule", topoproto.TabletAliasString(opts.NewPrimaryAlias))
+			}
+			continue
+		}
+		// If ERS is configured to prevent cross cell promotions, remove any tablet not from the same cell as the previous primary
+		if opts.PreventCrossCellPromotion && prevPrimary != nil && tablet.Alias.Cell != prevPrimary.Alias.Cell {
+			erp.logger.Infof("Removing %s from list of valid candidates for promotion because it isn't in the same cell as the previous primary", tabletAliasStr)
+			if opts.NewPrimaryAlias != nil && topoproto.TabletAliasEqual(opts.NewPrimaryAlias, tablet.Alias) {
+				return nil, vterrors.Errorf(vtrpc.Code_ABORTED, "proposed primary %s is is a different cell as the previous primary", topoproto.TabletAliasString(opts.NewPrimaryAlias))
+			}
+			continue
+		}
+		// Remove any tablet which cannot make forward progress using the list of tablets we have reached
+		if !canEstablishForTablet(tablet, tabletsReachable) {
+			erp.logger.Infof("Removing %s from list of valid candidates for promotion because it will not be able to make forward progress on promotion with the tablets currently reachable", tabletAliasStr)
+			if opts.NewPrimaryAlias != nil && topoproto.TabletAliasEqual(opts.NewPrimaryAlias, tablet.Alias) {
+				return nil, vterrors.Errorf(vtrpc.Code_ABORTED, "proposed primary %s will not be able to make forward progress on being promoted", topoproto.TabletAliasString(opts.NewPrimaryAlias))
+			}
+			continue
+		}
+		restrictedValidTablets = append(restrictedValidTablets, tablet)
+	}
+	return restrictedValidTablets, nil
+}
diff --git a/go/vt/vtctl/reparentutil/replication.go b/go/vt/vtctl/reparentutil/replication.go
index d7d7f1aee3..ebba68b6a1 100644
--- a/go/vt/vtctl/reparentutil/replication.go
+++ b/go/vt/vtctl/reparentutil/replication.go
@@ -151,10 +151,19 @@ func ReplicaWasRunning(stopStatus *replicationdatapb.StopReplicationStatus) (boo
 	return stopStatus.Before.IoThreadRunning || stopStatus.Before.SqlThreadRunning, nil
 }
 
-// StopReplicationAndBuildStatusMaps stops replication on all replicas, then
+// replicationSnapshot stores the status maps and the tablets that were reachable
+// when trying to stopReplicationAndBuildStatusMaps.
+type replicationSnapshot struct {
+	statusMap        map[string]*replicationdatapb.StopReplicationStatus
+	primaryStatusMap map[string]*replicationdatapb.PrimaryStatus
+	reachableTablets []*topodatapb.Tablet
+}
+
+// stopReplicationAndBuildStatusMaps stops replication on all replicas, then
 // collects and returns a mapping of TabletAlias (as string) to their current
 // replication positions.
-func StopReplicationAndBuildStatusMaps(
+// Apart from the status maps, it also returns the tablets reached as a list
+func stopReplicationAndBuildStatusMaps(
 	ctx context.Context,
 	tmc tmclient.TabletManagerClient,
 	ev *events.Reparent,
@@ -163,16 +172,18 @@ func StopReplicationAndBuildStatusMaps(
 	ignoredTablets sets.String,
 	tabletToWaitFor *topodatapb.TabletAlias,
 	logger logutil.Logger,
-) (map[string]*replicationdatapb.StopReplicationStatus, map[string]*replicationdatapb.PrimaryStatus, error) {
+) (*replicationSnapshot, error) {
 	event.DispatchUpdate(ev, "stop replication on all replicas")
 
 	var (
-		statusMap        = map[string]*replicationdatapb.StopReplicationStatus{}
-		primaryStatusMap = map[string]*replicationdatapb.PrimaryStatus{}
-		m                sync.Mutex
-		errChan          = make(chan concurrency.Error)
-		tabletsReachable []*topodatapb.Tablet
-		allTablets       []*topodatapb.Tablet
+		m          sync.Mutex
+		errChan    = make(chan concurrency.Error)
+		allTablets []*topodatapb.Tablet
+		res        = &replicationSnapshot{
+			statusMap:        map[string]*replicationdatapb.StopReplicationStatus{},
+			primaryStatusMap: map[string]*replicationdatapb.PrimaryStatus{},
+			reachableTablets: []*topodatapb.Tablet{},
+		}
 	)
 
 	groupCtx, groupCancel := context.WithTimeout(ctx, waitReplicasTimeout)
@@ -205,8 +216,8 @@ func StopReplicationAndBuildStatusMaps(
 				}
 
 				m.Lock()
-				primaryStatusMap[alias] = primaryStatus
-				tabletsReachable = append(tabletsReachable, tabletInfo.Tablet)
+				res.primaryStatusMap[alias] = primaryStatus
+				res.reachableTablets = append(res.reachableTablets, tabletInfo.Tablet)
 				m.Unlock()
 			} else {
 				logger.Warningf("failed to get replication status from %v: %v", alias, err)
@@ -214,8 +225,8 @@ func StopReplicationAndBuildStatusMaps(
 			}
 		} else {
 			m.Lock()
-			statusMap[alias] = stopReplicationStatus
-			tabletsReachable = append(tabletsReachable, tabletInfo.Tablet)
+			res.statusMap[alias] = stopReplicationStatus
+			res.reachableTablets = append(res.reachableTablets, tabletInfo.Tablet)
 			m.Unlock()
 		}
 	}
@@ -245,15 +256,15 @@ func StopReplicationAndBuildStatusMaps(
 
 	errRecorder := errgroup.Wait(groupCancel, errChan)
 	if len(errRecorder.Errors) <= 1 {
-		return statusMap, primaryStatusMap, nil
+		return res, nil
 	}
 	// check that the tablets we were able to reach are sufficient for us to guarantee that no new write will be accepted by any tablet
-	revokeSuccessful := haveRevoked(tabletsReachable, allTablets)
+	revokeSuccessful := haveRevoked(res.reachableTablets, allTablets)
 	if !revokeSuccessful {
-		return nil, nil, vterrors.Wrapf(errRecorder.Error(), "could not reach sufficient tablets to guarantee safety: %v", errRecorder.Error())
+		return nil, vterrors.Wrapf(errRecorder.Error(), "could not reach sufficient tablets to guarantee safety: %v", errRecorder.Error())
 	}
 
-	return statusMap, primaryStatusMap, nil
+	return res, nil
 }
 
 // WaitForRelayLogsToApply blocks execution waiting for the given tablet's relay
diff --git a/go/vt/vtgate/engine/filter.go b/go/vt/vtgate/engine/filter.go
index 79a028d298..ff6480fe77 100644
--- a/go/vt/vtgate/engine/filter.go
+++ b/go/vt/vtgate/engine/filter.go
@@ -59,6 +59,7 @@ func (f *Filter) TryExecute(vcursor VCursor, bindVars map[string]*querypb.BindVa
 	}
 	env := evalengine.EnvWithBindVars(bindVars, vcursor.ConnCollation())
 	var rows [][]sqltypes.Value
+	env.Fields = result.Fields
 	for _, row := range result.Rows {
 		env.Row = row
 		evalResult, err := env.Evaluate(f.Predicate)
@@ -82,6 +83,7 @@ func (f *Filter) TryStreamExecute(vcursor VCursor, bindVars map[string]*querypb.
 	env := evalengine.EnvWithBindVars(bindVars, vcursor.ConnCollation())
 	filter := func(results *sqltypes.Result) error {
 		var rows [][]sqltypes.Value
+		env.Fields = results.Fields
 		for _, row := range results.Rows {
 			env.Row = row
 			evalResult, err := env.Evaluate(f.Predicate)
diff --git a/go/vt/vtgate/engine/ordered_aggregate.go b/go/vt/vtgate/engine/ordered_aggregate.go
index 90aa1e6222..1c6b82d592 100644
--- a/go/vt/vtgate/engine/ordered_aggregate.go
+++ b/go/vt/vtgate/engine/ordered_aggregate.go
@@ -224,7 +224,7 @@ func (oa *OrderedAggregate) execute(vcursor VCursor, bindVars map[string]*queryp
 		return nil, err
 	}
 	out := &sqltypes.Result{
-		Fields: oa.convertFields(result.Fields),
+		Fields: convertFields(result.Fields, oa.PreProcess, oa.Aggregates),
 		Rows:   make([][]sqltypes.Value, 0, len(result.Rows)),
 	}
 	// This code is similar to the one in StreamExecute.
@@ -232,7 +232,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 = oa.convertRow(row)
+			current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates)
 			continue
 		}
 		equal, err := oa.keysEqual(current, row, oa.Collations)
@@ -241,28 +241,18 @@ func (oa *OrderedAggregate) execute(vcursor VCursor, bindVars map[string]*queryp
 		}
 
 		if equal {
-			current, curDistincts, err = oa.merge(result.Fields, current, row, curDistincts, oa.Collations)
+			current, curDistincts, err = merge(result.Fields, current, row, curDistincts, oa.Collations, oa.Aggregates)
 			if err != nil {
 				return nil, err
 			}
 			continue
 		}
 		out.Rows = append(out.Rows, current)
-		current, curDistincts = oa.convertRow(row)
-	}
-
-	if len(result.Rows) == 0 && len(oa.GroupByKeys) == 0 {
-		// When doing aggregation without grouping keys, we need to produce a single row containing zero-value for the
-		// different aggregation functions
-		row, err := oa.createEmptyRow()
-		if err != nil {
-			return nil, err
-		}
-		out.Rows = append(out.Rows, row)
+		current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates)
 	}
 
 	if current != nil {
-		final, err := oa.convertFinal(current)
+		final, err := convertFinal(current, oa.Aggregates)
 		if err != nil {
 			return nil, err
 		}
@@ -283,7 +273,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 = oa.convertFields(qr.Fields)
+			fields = convertFields(qr.Fields, oa.PreProcess, oa.Aggregates)
 			if err := cb(&sqltypes.Result{Fields: fields}); err != nil {
 				return err
 			}
@@ -291,7 +281,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 = oa.convertRow(row)
+				current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates)
 				continue
 			}
 
@@ -301,7 +291,7 @@ func (oa *OrderedAggregate) TryStreamExecute(vcursor VCursor, bindVars map[strin
 			}
 
 			if equal {
-				current, curDistincts, err = oa.merge(fields, current, row, curDistincts, oa.Collations)
+				current, curDistincts, err = merge(fields, current, row, curDistincts, oa.Collations, oa.Aggregates)
 				if err != nil {
 					return err
 				}
@@ -310,7 +300,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 = oa.convertRow(row)
+			current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates)
 		}
 		return nil
 	})
@@ -326,11 +316,11 @@ func (oa *OrderedAggregate) TryStreamExecute(vcursor VCursor, bindVars map[strin
 	return nil
 }
 
-func (oa *OrderedAggregate) convertFields(fields []*querypb.Field) []*querypb.Field {
-	if !oa.PreProcess {
+func convertFields(fields []*querypb.Field, preProcess bool, aggrs []*AggregateParams) []*querypb.Field {
+	if !preProcess {
 		return fields
 	}
-	for _, aggr := range oa.Aggregates {
+	for _, aggr := range aggrs {
 		if !aggr.preProcess() {
 			continue
 		}
@@ -345,13 +335,13 @@ func (oa *OrderedAggregate) convertFields(fields []*querypb.Field) []*querypb.Fi
 	return fields
 }
 
-func (oa *OrderedAggregate) convertRow(row []sqltypes.Value) (newRow []sqltypes.Value, curDistincts []sqltypes.Value) {
-	if !oa.PreProcess {
+func convertRow(row []sqltypes.Value, preProcess bool, aggregates []*AggregateParams) (newRow []sqltypes.Value, curDistincts []sqltypes.Value) {
+	if !preProcess {
 		return row, nil
 	}
 	newRow = append(newRow, row...)
-	curDistincts = make([]sqltypes.Value, len(oa.Aggregates))
-	for index, aggr := range oa.Aggregates {
+	curDistincts = make([]sqltypes.Value, len(aggregates))
+	for index, aggr := range aggregates {
 		switch aggr.Opcode {
 		case AggregateCountDistinct:
 			curDistincts[index] = findComparableCurrentDistinct(row, aggr)
@@ -398,7 +388,7 @@ func (oa *OrderedAggregate) GetFields(vcursor VCursor, bindVars map[string]*quer
 	if err != nil {
 		return nil, err
 	}
-	qr = &sqltypes.Result{Fields: oa.convertFields(qr.Fields)}
+	qr = &sqltypes.Result{Fields: convertFields(qr.Fields, oa.PreProcess, oa.Aggregates)}
 	return qr.Truncate(oa.TruncateColumnCount), nil
 }
 
@@ -434,9 +424,15 @@ func (oa *OrderedAggregate) keysEqual(row1, row2 []sqltypes.Value, colls map[int
 	return true, nil
 }
 
-func (oa *OrderedAggregate) merge(fields []*querypb.Field, row1, row2 []sqltypes.Value, curDistincts []sqltypes.Value, colls map[int]collations.ID) ([]sqltypes.Value, []sqltypes.Value, error) {
+func merge(
+	fields []*querypb.Field,
+	row1, row2 []sqltypes.Value,
+	curDistincts []sqltypes.Value,
+	colls map[int]collations.ID,
+	aggregates []*AggregateParams,
+) ([]sqltypes.Value, []sqltypes.Value, error) {
 	result := sqltypes.CopyRow(row1)
-	for index, aggr := range oa.Aggregates {
+	for index, aggr := range aggregates {
 		if aggr.isDistinct() {
 			if row2[aggr.KeyCol].IsNull() {
 				continue
@@ -492,19 +488,6 @@ func (oa *OrderedAggregate) merge(fields []*querypb.Field, row1, row2 []sqltypes
 	return result, curDistincts, nil
 }
 
-// creates the empty row for the case when we are missing grouping keys and have empty input table
-func (oa *OrderedAggregate) createEmptyRow() ([]sqltypes.Value, error) {
-	out := make([]sqltypes.Value, len(oa.Aggregates))
-	for i, aggr := range oa.Aggregates {
-		value, err := createEmptyValueFor(aggr.Opcode)
-		if err != nil {
-			return nil, err
-		}
-		out[i] = value
-	}
-	return out, nil
-}
-
 func createEmptyValueFor(opcode AggregateOpcode) (sqltypes.Value, error) {
 	switch opcode {
 	case
@@ -547,9 +530,9 @@ func (oa *OrderedAggregate) description() PrimitiveDescription {
 	}
 }
 
-func (oa *OrderedAggregate) convertFinal(current []sqltypes.Value) ([]sqltypes.Value, error) {
+func convertFinal(current []sqltypes.Value, aggregates []*AggregateParams) ([]sqltypes.Value, error) {
 	result := sqltypes.CopyRow(current)
-	for _, aggr := range oa.Aggregates {
+	for _, aggr := range aggregates {
 		switch aggr.Opcode {
 		case AggregateGtid:
 			vgtid := &binlogdatapb.VGtid{}
diff --git a/go/vt/vtgate/engine/projection.go b/go/vt/vtgate/engine/projection.go
index ecb38d3d9b..81d8db4e94 100644
--- a/go/vt/vtgate/engine/projection.go
+++ b/go/vt/vtgate/engine/projection.go
@@ -39,23 +39,25 @@ func (p *Projection) TryExecute(vcursor VCursor, bindVars map[string]*querypb.Bi
 	}
 
 	env := evalengine.EnvWithBindVars(bindVars, vcursor.ConnCollation())
-	if wantfields {
-		err := p.addFields(env, result)
-		if err != nil {
-			return nil, err
-		}
-	}
+	env.Fields = result.Fields
 	var rows [][]sqltypes.Value
 	for _, row := range result.Rows {
 		env.Row = row
+		resRow := make([]sqltypes.Value, 0, len(p.Exprs))
 		for _, exp := range p.Exprs {
 			result, err := env.Evaluate(exp)
 			if err != nil {
 				return nil, err
 			}
-			row = append(row, result.Value())
+			resRow = append(resRow, result.Value())
+		}
+		rows = append(rows, resRow)
+	}
+	if wantfields {
+		err := p.addFields(env, result)
+		if err != nil {
+			return nil, err
 		}
-		rows = append(rows, row)
 	}
 	result.Rows = rows
 	return result, nil
@@ -76,6 +78,7 @@ func (p *Projection) TryStreamExecute(vcursor VCursor, bindVars map[string]*quer
 		}
 	}
 	var rows [][]sqltypes.Value
+	env.Fields = result.Fields
 	for _, row := range result.Rows {
 		env.Row = row
 		for _, exp := range p.Exprs {
@@ -106,6 +109,7 @@ func (p *Projection) GetFields(vcursor VCursor, bindVars map[string]*querypb.Bin
 }
 
 func (p *Projection) addFields(env *evalengine.ExpressionEnv, qr *sqltypes.Result) error {
+	qr.Fields = nil
 	for i, col := range p.Cols {
 		q, err := env.TypeOf(p.Exprs[i])
 		if err != nil {
diff --git a/go/vt/vtgate/engine/scalar_aggregation.go b/go/vt/vtgate/engine/scalar_aggregation.go
new file mode 100644
index 0000000000..10884ae5f8
--- /dev/null
+++ b/go/vt/vtgate/engine/scalar_aggregation.go
@@ -0,0 +1,193 @@
+/*
+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 engine
+
+import (
+	"vitess.io/vitess/go/mysql/collations"
+	"vitess.io/vitess/go/sqltypes"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+)
+
+var _ Primitive = (*ScalarAggregate)(nil)
+
+// ScalarAggregate is a primitive used to do aggregations without grouping keys
+type ScalarAggregate 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
+
+	// TruncateColumnCount specifies the number of columns to return
+	// in the final result. Rest of the columns are truncated
+	// from the result received. If 0, no truncation happens.
+	TruncateColumnCount int `json:",omitempty"`
+
+	// Collations stores the collation ID per column offset.
+	// It is used for grouping keys and distinct aggregate functions
+	Collations map[int]collations.ID
+
+	// Input is the primitive that will feed into this Primitive.
+	Input Primitive
+}
+
+// RouteType implements the Primitive interface
+func (sa *ScalarAggregate) RouteType() string {
+	return sa.Input.RouteType()
+}
+
+// GetKeyspaceName implements the Primitive interface
+func (sa *ScalarAggregate) GetKeyspaceName() string {
+	return sa.Input.GetKeyspaceName()
+
+}
+
+// GetTableName implements the Primitive interface
+func (sa *ScalarAggregate) GetTableName() string {
+	return sa.Input.GetTableName()
+}
+
+// GetFields implements the Primitive interface
+func (sa *ScalarAggregate) GetFields(vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
+	qr, err := sa.Input.GetFields(vcursor, bindVars)
+	if err != nil {
+		return nil, err
+	}
+	qr = &sqltypes.Result{Fields: convertFields(qr.Fields, sa.PreProcess, sa.Aggregates)}
+	return qr.Truncate(sa.TruncateColumnCount), nil
+}
+
+// NeedsTransaction implements the Primitive interface
+func (sa *ScalarAggregate) NeedsTransaction() bool {
+	return sa.Input.NeedsTransaction()
+}
+
+// TryExecute implements the Primitive interface
+func (sa *ScalarAggregate) TryExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool) (*sqltypes.Result, error) {
+	result, err := vcursor.ExecutePrimitive(sa.Input, bindVars, wantfields)
+	if err != nil {
+		return nil, err
+	}
+	out := &sqltypes.Result{
+		Fields: convertFields(result.Fields, sa.PreProcess, sa.Aggregates),
+	}
+
+	var resultRow []sqltypes.Value
+	var curDistincts []sqltypes.Value
+	for _, row := range result.Rows {
+		if resultRow == nil {
+			resultRow, curDistincts = convertRow(row, sa.PreProcess, sa.Aggregates)
+			continue
+		}
+		resultRow, curDistincts, err = merge(result.Fields, resultRow, row, curDistincts, sa.Collations, sa.Aggregates)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if resultRow == nil {
+		// When doing aggregation without grouping keys, we need to produce a single row containing zero-value for the
+		// different aggregation functions
+		resultRow, err = sa.createEmptyRow()
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		resultRow, err = convertFinal(resultRow, sa.Aggregates)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	out.Rows = [][]sqltypes.Value{resultRow}
+	return out, nil
+}
+
+// TryStreamExecute implements the Primitive interface
+func (sa *ScalarAggregate) TryStreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
+	cb := func(qr *sqltypes.Result) error {
+		return callback(qr.Truncate(sa.TruncateColumnCount))
+	}
+	var current []sqltypes.Value
+	var curDistincts []sqltypes.Value
+	var fields []*querypb.Field
+	fieldsSent := false
+
+	err := vcursor.StreamExecutePrimitive(sa.Input, bindVars, wantfields, func(result *sqltypes.Result) error {
+		if len(result.Fields) != 0 && !fieldsSent {
+			fields = convertFields(result.Fields, sa.PreProcess, sa.Aggregates)
+			if err := cb(&sqltypes.Result{Fields: fields}); err != nil {
+				return err
+			}
+			fieldsSent = true
+		}
+
+		// 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)
+				continue
+			}
+			var err error
+			current, curDistincts, err = merge(fields, current, row, curDistincts, sa.Collations, sa.Aggregates)
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	return cb(&sqltypes.Result{Rows: [][]sqltypes.Value{current}})
+}
+
+// creates the empty row for the case when we are missing grouping keys and have empty input table
+func (sa *ScalarAggregate) createEmptyRow() ([]sqltypes.Value, error) {
+	out := make([]sqltypes.Value, len(sa.Aggregates))
+	for i, aggr := range sa.Aggregates {
+		value, err := createEmptyValueFor(aggr.Opcode)
+		if err != nil {
+			return nil, err
+		}
+		out[i] = value
+	}
+	return out, nil
+}
+
+// Inputs implements the Primitive interface
+func (sa *ScalarAggregate) Inputs() []Primitive {
+	return []Primitive{sa.Input}
+}
+
+// description implements the Primitive interface
+func (sa *ScalarAggregate) description() PrimitiveDescription {
+	aggregates := GenericJoin(sa.Aggregates, aggregateParamsToString)
+	other := map[string]interface{}{
+		"Aggregates": aggregates,
+	}
+	if sa.TruncateColumnCount > 0 {
+		other["ResultColumns"] = sa.TruncateColumnCount
+	}
+	return PrimitiveDescription{
+		OperatorType: "Aggregate",
+		Variant:      "Scalar",
+		Other:        other,
+	}
+
+}
diff --git a/go/vt/vtgate/engine/set.go b/go/vt/vtgate/engine/set.go
index 4c4d0adb18..1b8fb3a04e 100644
--- a/go/vt/vtgate/engine/set.go
+++ b/go/vt/vtgate/engine/set.go
@@ -123,6 +123,7 @@ func (s *Set) TryExecute(vcursor VCursor, bindVars map[string]*querypb.BindVaria
 	}
 	env := evalengine.EnvWithBindVars(bindVars, vcursor.ConnCollation())
 	env.Row = input.Rows[0]
+	env.Fields = input.Fields
 	for _, setOp := range s.Ops {
 		err := setOp.Execute(vcursor, env)
 		if err != nil {
diff --git a/go/vt/vtgate/evalengine/expressions.go b/go/vt/vtgate/evalengine/expressions.go
index 973e19d496..6cf856bd69 100644
--- a/go/vt/vtgate/evalengine/expressions.go
+++ b/go/vt/vtgate/evalengine/expressions.go
@@ -36,8 +36,11 @@ type (
 	// evaluates in, such as the current row and bindvars
 	ExpressionEnv struct {
 		BindVars         map[string]*querypb.BindVariable
-		Row              []sqltypes.Value
 		DefaultCollation collations.ID
+
+		// Row and Fields should line up
+		Row    []sqltypes.Value
+		Fields []*querypb.Field
 	}
 
 	// Expr is the interface that all evaluating expressions must implement
@@ -520,12 +523,18 @@ func (t TupleExpr) typeof(*ExpressionEnv) (sqltypes.Type, flag) {
 }
 
 func (c *Column) typeof(env *ExpressionEnv) (sqltypes.Type, flag) {
-	value := env.Row[c.Offset]
-	tt := value.Type()
-	switch tt {
-	case sqltypes.Null:
-		return tt, flagNullable | flagNull
-	default:
-		return tt, 0
+	// we'll try to do the best possible with the information we have
+	if c.Offset < len(env.Row) {
+		value := env.Row[c.Offset]
+		if value.IsNull() {
+			return sqltypes.Null, flagNull | flagNullable
+		}
+		return value.Type(), flag(0)
 	}
+
+	if c.Offset < len(env.Fields) {
+		return env.Fields[c.Offset].Type, flagNullable
+	}
+
+	panic("Column missing both data and field")
 }
diff --git a/go/vt/vtgate/planbuilder/grouping.go b/go/vt/vtgate/planbuilder/grouping.go
index 84335ebc61..2bf58dcfa4 100644
--- a/go/vt/vtgate/planbuilder/grouping.go
+++ b/go/vt/vtgate/planbuilder/grouping.go
@@ -77,7 +77,7 @@ func planGroupBy(pb *primitiveBuilder, input logicalPlan, groupBy sqlparser.Grou
 			default:
 				return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: in scatter query: only simple references allowed")
 			}
-			node.eaggr.GroupByKeys = append(node.eaggr.GroupByKeys, &engine.GroupByParams{KeyCol: colNumber, WeightStringCol: -1, FromGroupBy: true})
+			node.groupByKeys = append(node.groupByKeys, &engine.GroupByParams{KeyCol: colNumber, WeightStringCol: -1, FromGroupBy: true})
 		}
 		// Append the distinct aggregate if any.
 		if node.extraDistinct != nil {
@@ -110,7 +110,7 @@ func planDistinct(input logicalPlan) (logicalPlan, error) {
 			if rc.column.Origin() == node {
 				return newDistinct(node, nil), nil
 			}
-			node.eaggr.GroupByKeys = append(node.eaggr.GroupByKeys, &engine.GroupByParams{KeyCol: i, WeightStringCol: -1, FromGroupBy: false})
+			node.groupByKeys = append(node.groupByKeys, &engine.GroupByParams{KeyCol: i, WeightStringCol: -1, FromGroupBy: false})
 		}
 		newInput, err := planDistinct(node.input)
 		if err != nil {
diff --git a/go/vt/vtgate/planbuilder/horizon_planning.go b/go/vt/vtgate/planbuilder/horizon_planning.go
index d12cd7bcfa..aedd72c683 100644
--- a/go/vt/vtgate/planbuilder/horizon_planning.go
+++ b/go/vt/vtgate/planbuilder/horizon_planning.go
@@ -145,7 +145,7 @@ func (hp *horizonPlanning) truncateColumnsIfNeeded(ctx *plancontext.PlanningCont
 	case *joinGen4, *semiJoin, *hashJoin:
 		// since this is a join, we can safely add extra columns and not need to truncate them
 	case *orderedAggregate:
-		p.eaggr.SetTruncateColumnCount(hp.sel.GetColumnCount())
+		p.truncateColumnCount = hp.sel.GetColumnCount()
 	case *memorySort:
 		p.truncater.SetTruncateColumnCount(hp.sel.GetColumnCount())
 	case *pulloutSubquery:
@@ -336,7 +336,7 @@ func pushProjection(ctx *plancontext.PlanningContext, expr *sqlparser.AliasedExp
 		return len(node.eSimpleProj.Cols) - 1, true, nil
 	case *orderedAggregate:
 		colName, isColName := expr.Expr.(*sqlparser.ColName)
-		for _, aggregate := range node.eaggr.Aggregates {
+		for _, aggregate := range node.aggregates {
 			if sqlparser.EqualsExpr(aggregate.Expr, expr.Expr) {
 				return aggregate.Col, false, nil
 			}
@@ -344,7 +344,7 @@ func pushProjection(ctx *plancontext.PlanningContext, expr *sqlparser.AliasedExp
 				return aggregate.Col, false, nil
 			}
 		}
-		for _, key := range node.eaggr.GroupByKeys {
+		for _, key := range node.groupByKeys {
 			if sqlparser.EqualsExpr(key.Expr, expr.Expr) {
 				return key.KeyCol, false, nil
 			}
@@ -473,14 +473,11 @@ func (hp *horizonPlanning) planAggregations(ctx *plancontext.PlanningContext, pl
 		if hp.qp.ProjectionError != nil {
 			return nil, hp.qp.ProjectionError
 		}
-		eaggr := &engine.OrderedAggregate{}
 		oa = &orderedAggregate{
 			resultsBuilder: resultsBuilder{
 				logicalPlanCommon: newBuilderCommon(plan),
 				weightStrings:     make(map[*resultColumn]int),
-				truncater:         eaggr,
 			},
-			eaggr: eaggr,
 		}
 		newPlan = oa
 		hp.vtgateGrouping = true
@@ -526,7 +523,7 @@ func (hp *horizonPlanning) planAggregations(ctx *plancontext.PlanningContext, pl
 		}
 		param.Col = offset
 		param.Expr = fExpr
-		oa.eaggr.Aggregates = append(oa.eaggr.Aggregates, param)
+		oa.aggregates = append(oa.aggregates, param)
 	}
 
 	for _, groupExpr := range hp.qp.GroupByExprs {
@@ -580,7 +577,7 @@ func (hp *horizonPlanning) planAggregations(ctx *plancontext.PlanningContext, pl
 }
 
 // createPushExprAndAlias creates the expression that should be pushed down to the leaves,
-// and changes the opcode so it is a distinct one if needed
+// and changes the opcode, so it is a distinct one if needed
 func (hp *horizonPlanning) createPushExprAndAlias(
 	ctx *plancontext.PlanningContext,
 	expr abstract.SelectExpr,
@@ -609,11 +606,11 @@ func (hp *horizonPlanning) createPushExprAndAlias(
 			opcode = engine.AggregateSumDistinct
 		}
 
-		oa.eaggr.PreProcess = true
+		oa.preProcess = true
 		by := abstract.GroupBy{
 			Inner:             innerAliased.Expr,
 			WeightStrExpr:     innerAliased.Expr,
-			DistinctAggrIndex: len(oa.eaggr.Aggregates) + 1,
+			DistinctAggrIndex: len(oa.aggregates) + 1,
 		}
 		hp.qp.GroupByExprs = append(hp.qp.GroupByExprs, by)
 	}
@@ -660,11 +657,11 @@ func planGroupByGen4(ctx *plancontext.PlanningContext, groupExpr abstract.GroupB
 			return err
 		}
 		if groupExpr.DistinctAggrIndex == 0 {
-			node.eaggr.GroupByKeys = append(node.eaggr.GroupByKeys, &engine.GroupByParams{KeyCol: keyCol, WeightStringCol: wsOffset, Expr: groupExpr.WeightStrExpr, CollationID: ctx.SemTable.CollationForExpr(groupExpr.Inner)})
+			node.groupByKeys = append(node.groupByKeys, &engine.GroupByParams{KeyCol: keyCol, WeightStringCol: wsOffset, Expr: groupExpr.WeightStrExpr, CollationID: ctx.SemTable.CollationForExpr(groupExpr.Inner)})
 		} else {
 			if wsOffset != -1 {
-				node.eaggr.Aggregates[groupExpr.DistinctAggrIndex-1].WAssigned = true
-				node.eaggr.Aggregates[groupExpr.DistinctAggrIndex-1].WCol = wsOffset
+				node.aggregates[groupExpr.DistinctAggrIndex-1].WAssigned = true
+				node.aggregates[groupExpr.DistinctAggrIndex-1].WCol = wsOffset
 			}
 		}
 		err = planGroupByGen4(ctx, groupExpr, node.input, wsOffset != -1)
@@ -949,7 +946,7 @@ func createMemorySortPlanOnAggregation(plan *orderedAggregate, orderExprs []abst
 
 		collationID := collations.Unknown
 		if woffset != -1 {
-			collationID = plan.eaggr.GroupByKeys[idx].CollationID
+			collationID = plan.groupByKeys[idx].CollationID
 		}
 		ms.eMemorySort.OrderBy = append(ms.eMemorySort.OrderBy, engine.OrderByParams{
 			Col:               offset,
@@ -963,12 +960,12 @@ func createMemorySortPlanOnAggregation(plan *orderedAggregate, orderExprs []abst
 }
 
 func findExprInOrderedAggr(plan *orderedAggregate, order abstract.OrderBy) (keyCol int, weightStringCol int, index int, found bool) {
-	for idx, key := range plan.eaggr.GroupByKeys {
+	for idx, key := range plan.groupByKeys {
 		if sqlparser.EqualsExpr(order.WeightStrExpr, key.Expr) {
 			return key.KeyCol, key.WeightStringCol, idx, true
 		}
 	}
-	for idx, aggregate := range plan.eaggr.Aggregates {
+	for idx, aggregate := range plan.aggregates {
 		if sqlparser.EqualsExpr(order.WeightStrExpr, aggregate.Expr) {
 			return aggregate.Col, -1, idx, true
 		}
@@ -1041,14 +1038,11 @@ func (hp *horizonPlanning) planDistinct(ctx *plancontext.PlanningContext, plan l
 }
 
 func (hp *horizonPlanning) planDistinctOA(semTable *semantics.SemTable, currPlan *orderedAggregate) (logicalPlan, error) {
-	eaggr := &engine.OrderedAggregate{}
 	oa := &orderedAggregate{
 		resultsBuilder: resultsBuilder{
 			logicalPlanCommon: newBuilderCommon(currPlan),
 			weightStrings:     make(map[*resultColumn]int),
-			truncater:         eaggr,
 		},
-		eaggr: eaggr,
 	}
 	for _, sExpr := range hp.qp.SelectExprs {
 		expr, err := sExpr.GetExpr()
@@ -1056,20 +1050,20 @@ func (hp *horizonPlanning) planDistinctOA(semTable *semantics.SemTable, currPlan
 			return nil, err
 		}
 		found := false
-		for _, grpParam := range currPlan.eaggr.GroupByKeys {
+		for _, grpParam := range currPlan.groupByKeys {
 			if sqlparser.EqualsExpr(expr, grpParam.Expr) {
 				found = true
-				eaggr.GroupByKeys = append(eaggr.GroupByKeys, grpParam)
+				oa.groupByKeys = append(oa.groupByKeys, grpParam)
 				break
 			}
 		}
 		if found {
 			continue
 		}
-		for _, aggrParam := range currPlan.eaggr.Aggregates {
+		for _, aggrParam := range currPlan.aggregates {
 			if sqlparser.EqualsExpr(expr, aggrParam.Expr) {
 				found = true
-				eaggr.GroupByKeys = append(eaggr.GroupByKeys, &engine.GroupByParams{KeyCol: aggrParam.Col, WeightStringCol: -1, CollationID: semTable.CollationForExpr(expr)})
+				oa.groupByKeys = append(oa.groupByKeys, &engine.GroupByParams{KeyCol: aggrParam.Col, WeightStringCol: -1, CollationID: semTable.CollationForExpr(expr)})
 				break
 			}
 		}
@@ -1081,8 +1075,8 @@ func (hp *horizonPlanning) planDistinctOA(semTable *semantics.SemTable, currPlan
 }
 
 func (hp *horizonPlanning) addDistinct(ctx *plancontext.PlanningContext, plan logicalPlan) (logicalPlan, error) {
-	eaggr := &engine.OrderedAggregate{}
 	var orderExprs []abstract.OrderBy
+	var groupByKeys []*engine.GroupByParams
 	for index, sExpr := range hp.qp.SelectExprs {
 		aliasExpr, err := sExpr.GetAliasedExpr()
 		if err != nil {
@@ -1107,7 +1101,7 @@ func (hp *horizonPlanning) addDistinct(ctx *plancontext.PlanningContext, plan lo
 			return nil, err
 		}
 		grpParam.WeightStringCol = wOffset
-		eaggr.GroupByKeys = append(eaggr.GroupByKeys, grpParam)
+		groupByKeys = append(groupByKeys, grpParam)
 
 		orderExprs = append(orderExprs, abstract.OrderBy{
 			Inner:         &sqlparser.Order{Expr: inner},
@@ -1122,9 +1116,8 @@ func (hp *horizonPlanning) addDistinct(ctx *plancontext.PlanningContext, plan lo
 		resultsBuilder: resultsBuilder{
 			logicalPlanCommon: newBuilderCommon(innerPlan),
 			weightStrings:     make(map[*resultColumn]int),
-			truncater:         eaggr,
 		},
-		eaggr: eaggr,
+		groupByKeys: groupByKeys,
 	}
 	return oa, nil
 }
diff --git a/go/vt/vtgate/planbuilder/ordered_aggregate.go b/go/vt/vtgate/planbuilder/ordered_aggregate.go
index 8cc037d6ac..de1fff1c97 100644
--- a/go/vt/vtgate/planbuilder/ordered_aggregate.go
+++ b/go/vt/vtgate/planbuilder/ordered_aggregate.go
@@ -63,7 +63,19 @@ var _ logicalPlan = (*orderedAggregate)(nil)
 type orderedAggregate struct {
 	resultsBuilder
 	extraDistinct *sqlparser.ColName
-	eaggr         *engine.OrderedAggregate
+
+	// preProcess is true if one of the aggregates needs preprocessing.
+	preProcess bool
+
+	// aggregates specifies the aggregation parameters for each
+	// aggregation function: function opcode and input column number.
+	aggregates []*engine.AggregateParams
+
+	// groupByKeys specifies the input values that must be used for
+	// the aggregation key.
+	groupByKeys []*engine.GroupByParams
+
+	truncateColumnCount int
 }
 
 // checkAggregates analyzes the select expression for aggregates. If it determines
@@ -129,11 +141,9 @@ func (pb *primitiveBuilder) checkAggregates(sel *sqlparser.Select) error {
 	}
 
 	// We need an aggregator primitive.
-	eaggr := &engine.OrderedAggregate{}
-	pb.plan = &orderedAggregate{
-		resultsBuilder: newResultsBuilder(rb, eaggr),
-		eaggr:          eaggr,
-	}
+	oa := &orderedAggregate{}
+	oa.resultsBuilder = newResultsBuilder(rb, oa)
+	pb.plan = oa
 	pb.plan.Reorder(0)
 	return nil
 }
@@ -215,8 +225,36 @@ func findAlias(colname *sqlparser.ColName, selects sqlparser.SelectExprs) sqlpar
 
 // Primitive implements the logicalPlan interface
 func (oa *orderedAggregate) Primitive() engine.Primitive {
-	oa.eaggr.Input = oa.input.Primitive()
-	return oa.eaggr
+	colls := map[int]collations.ID{}
+	for _, key := range oa.aggregates {
+		if key.CollationID != collations.Unknown {
+			colls[key.KeyCol] = key.CollationID
+		}
+	}
+	for _, key := range oa.groupByKeys {
+		if key.CollationID != collations.Unknown {
+			colls[key.KeyCol] = key.CollationID
+		}
+	}
+
+	input := oa.input.Primitive()
+	if len(oa.groupByKeys) == 0 {
+		return &engine.ScalarAggregate{
+			PreProcess:          oa.preProcess,
+			Aggregates:          oa.aggregates,
+			TruncateColumnCount: oa.truncateColumnCount,
+			Collations:          colls,
+			Input:               input,
+		}
+	}
+
+	return &engine.OrderedAggregate{
+		PreProcess:          oa.preProcess,
+		Aggregates:          oa.aggregates,
+		GroupByKeys:         oa.groupByKeys,
+		TruncateColumnCount: oa.truncateColumnCount,
+		Input:               input,
+	}
 }
 
 func (oa *orderedAggregate) pushAggr(pb *primitiveBuilder, expr *sqlparser.AliasedExpr, origin logicalPlan) (rc *resultColumn, colNumber int, err error) {
@@ -245,7 +283,7 @@ func (oa *orderedAggregate) pushAggr(pb *primitiveBuilder, expr *sqlparser.Alias
 			return nil, 0, err
 		}
 		oa.extraDistinct = col
-		oa.eaggr.PreProcess = true
+		oa.preProcess = true
 		var alias string
 		if expr.As.IsEmpty() {
 			alias = sqlparser.String(expr.Expr)
@@ -258,7 +296,7 @@ func (oa *orderedAggregate) pushAggr(pb *primitiveBuilder, expr *sqlparser.Alias
 		case engine.AggregateSum:
 			opcode = engine.AggregateSumDistinct
 		}
-		oa.eaggr.Aggregates = append(oa.eaggr.Aggregates, &engine.AggregateParams{
+		oa.aggregates = append(oa.aggregates, &engine.AggregateParams{
 			Opcode: opcode,
 			Col:    innerCol,
 			Alias:  alias,
@@ -269,7 +307,7 @@ func (oa *orderedAggregate) pushAggr(pb *primitiveBuilder, expr *sqlparser.Alias
 			return nil, 0, err
 		}
 		pb.plan = newBuilder
-		oa.eaggr.Aggregates = append(oa.eaggr.Aggregates, &engine.AggregateParams{
+		oa.aggregates = append(oa.aggregates, &engine.AggregateParams{
 			Opcode: opcode,
 			Col:    innerCol,
 		})
@@ -314,7 +352,7 @@ func (oa *orderedAggregate) needDistinctHandling(pb *primitiveBuilder, funcExpr
 // compare those instead. This is because we currently don't have the
 // ability to mimic mysql's collation behavior.
 func (oa *orderedAggregate) Wireup(plan logicalPlan, jt *jointab) error {
-	for i, gbk := range oa.eaggr.GroupByKeys {
+	for i, gbk := range oa.groupByKeys {
 		rc := oa.resultColumns[gbk.KeyCol]
 		if sqltypes.IsText(rc.column.typ) {
 			weightcolNumber, err := oa.input.SupplyWeightString(gbk.KeyCol, gbk.FromGroupBy)
@@ -326,38 +364,31 @@ func (oa *orderedAggregate) Wireup(plan logicalPlan, jt *jointab) error {
 				return err
 			}
 			oa.weightStrings[rc] = weightcolNumber
-			oa.eaggr.GroupByKeys[i].WeightStringCol = weightcolNumber
-			oa.eaggr.GroupByKeys[i].KeyCol = weightcolNumber
-			oa.eaggr.TruncateColumnCount = len(oa.resultColumns)
+			oa.groupByKeys[i].WeightStringCol = weightcolNumber
+			oa.groupByKeys[i].KeyCol = weightcolNumber
+			oa.truncateColumnCount = len(oa.resultColumns)
 		}
 	}
 	return oa.input.Wireup(plan, jt)
 }
 
 func (oa *orderedAggregate) WireupGen4(semTable *semantics.SemTable) error {
-	colls := map[int]collations.ID{}
-	oa.eaggr.Collations = colls
-	for _, key := range oa.eaggr.Aggregates {
-		if key.CollationID != collations.Unknown {
-			colls[key.KeyCol] = key.CollationID
-		}
-	}
-	for _, key := range oa.eaggr.GroupByKeys {
-		if key.CollationID != collations.Unknown {
-			colls[key.KeyCol] = key.CollationID
-		}
-	}
 	return oa.input.WireupGen4(semTable)
 }
 
 // OutputColumns implements the logicalPlan interface
 func (oa *orderedAggregate) OutputColumns() []sqlparser.SelectExpr {
 	outputCols := sqlparser.CloneSelectExprs(oa.input.OutputColumns())
-	for _, aggr := range oa.eaggr.Aggregates {
+	for _, aggr := range oa.aggregates {
 		outputCols[aggr.Col] = &sqlparser.AliasedExpr{Expr: aggr.Expr, As: sqlparser.NewColIdent(aggr.Alias)}
 	}
-	if oa.eaggr.TruncateColumnCount > 0 {
-		return outputCols[:oa.eaggr.TruncateColumnCount]
+	if oa.truncateColumnCount > 0 {
+		return outputCols[:oa.truncateColumnCount]
 	}
 	return outputCols
 }
+
+// SetTruncateColumnCount sets the truncate column count.
+func (oa *orderedAggregate) SetTruncateColumnCount(count int) {
+	oa.truncateColumnCount = count
+}
diff --git a/go/vt/vtgate/planbuilder/ordering.go b/go/vt/vtgate/planbuilder/ordering.go
index db711f0e72..778316bb16 100644
--- a/go/vt/vtgate/planbuilder/ordering.go
+++ b/go/vt/vtgate/planbuilder/ordering.go
@@ -86,7 +86,7 @@ func planOAOrdering(pb *primitiveBuilder, orderBy v3OrderBy, oa *orderedAggregat
 	}
 
 	// referenced tracks the keys referenced by the order by clause.
-	referenced := make([]bool, len(oa.eaggr.GroupByKeys))
+	referenced := make([]bool, len(oa.groupByKeys))
 	postSort := false
 	selOrderBy := make(v3OrderBy, 0, len(orderBy))
 	for _, order := range orderBy {
@@ -113,7 +113,7 @@ func planOAOrdering(pb *primitiveBuilder, orderBy v3OrderBy, oa *orderedAggregat
 
 		// Match orderByCol against the group by columns.
 		found := false
-		for j, groupBy := range oa.eaggr.GroupByKeys {
+		for j, groupBy := range oa.groupByKeys {
 			if oa.resultColumns[groupBy.KeyCol].column != orderByCol {
 				continue
 			}
@@ -130,7 +130,7 @@ func planOAOrdering(pb *primitiveBuilder, orderBy v3OrderBy, oa *orderedAggregat
 	}
 
 	// Append any unreferenced keys at the end of the order by.
-	for i, groupByKey := range oa.eaggr.GroupByKeys {
+	for i, groupByKey := range oa.groupByKeys {
 		if referenced[i] {
 			continue
 		}
diff --git a/go/vt/vttablet/onlineddl/executor.go b/go/vt/vttablet/onlineddl/executor.go
index 7f1ba0dc3d..3b1b2d9bce 100644
--- a/go/vt/vttablet/onlineddl/executor.go
+++ b/go/vt/vttablet/onlineddl/executor.go
@@ -118,6 +118,8 @@ const (
 	etaSecondsUnknown                        = -1
 	etaSecondsNow                            = 0
 	rowsCopiedUnknown                        = 0
+	emptyHint                                = ""
+	readyToCompleteHint                      = "ready_to_complete"
 	databasePoolSize                         = 3
 	vreplicationCutOverThreshold             = 5 * time.Second
 	vreplicationTestSuiteWaitSeconds         = 5
@@ -548,7 +550,7 @@ func (e *Executor) executeDirectly(ctx context.Context, onlineDDL *schema.Online
 		return false, err
 	}
 
-	_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusRunning, false, progressPctStarted, etaSecondsUnknown, rowsCopiedUnknown)
+	_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusRunning, false, progressPctStarted, etaSecondsUnknown, rowsCopiedUnknown, emptyHint)
 	_, err = conn.ExecuteFetch(onlineDDL.SQL, 0, false)
 
 	if err != nil {
@@ -567,7 +569,7 @@ func (e *Executor) executeDirectly(ctx context.Context, onlineDDL *schema.Online
 	if err != nil {
 		return false, err
 	}
-	_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown)
+	_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown, emptyHint)
 
 	return acceptableErrorCodeFound, nil
 }
@@ -785,7 +787,7 @@ func (e *Executor) cutOverVReplMigration(ctx context.Context, s *VReplStream) er
 
 	// Tables are now swapped! Migration is successful
 	reenableWritesOnce() // this function is also deferred, in case of early return; but now would be a good time to resume writes, before we publish the migration as "complete"
-	_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, s.rowsCopied)
+	_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, s.rowsCopied, emptyHint)
 	return nil
 
 	// deferred function will re-enable writes now
@@ -928,7 +930,7 @@ func (e *Executor) ExecuteWithVReplication(ctx context.Context, onlineDDL *schem
 	defer conn.Close()
 
 	e.ownedRunningMigrations.Store(onlineDDL.UUID, onlineDDL)
-	if err := e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusRunning, false, progressPctStarted, etaSecondsUnknown, rowsCopiedUnknown); err != nil {
+	if err := e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusRunning, false, progressPctStarted, etaSecondsUnknown, rowsCopiedUnknown, emptyHint); err != nil {
 		return err
 	}
 
@@ -1075,24 +1077,28 @@ exit $exit_code
 		log.Errorf("Error creating wrapper script: %+v", err)
 		return err
 	}
-	onHookContent := func(status schema.OnlineDDLStatus) string {
+	onHookContent := func(status schema.OnlineDDLStatus, hint string) string {
 		return fmt.Sprintf(`#!/bin/bash
-	curl --max-time 10 -s 'http://localhost:%d/schema-migration/report-status?uuid=%s&status=%s&dryrun='"$GH_OST_DRY_RUN"'&progress='"$GH_OST_PROGRESS"'&eta='"$GH_OST_ETA_SECONDS"'&rowscopied='"$GH_OST_COPIED_ROWS"
-			`, *servenv.Port, onlineDDL.UUID, string(status))
+	curl --max-time 10 -s 'http://localhost:%d/schema-migration/report-status?uuid=%s&status=%s&hint=%s&dryrun='"$GH_OST_DRY_RUN"'&progress='"$GH_OST_PROGRESS"'&eta='"$GH_OST_ETA_SECONDS"'&rowscopied='"$GH_OST_COPIED_ROWS"
+			`, *servenv.Port, onlineDDL.UUID, string(status), hint)
 	}
-	if _, err := createTempScript(tempDir, "gh-ost-on-startup", onHookContent(schema.OnlineDDLStatusRunning)); err != nil {
+	if _, err := createTempScript(tempDir, "gh-ost-on-startup", onHookContent(schema.OnlineDDLStatusRunning, emptyHint)); err != nil {
 		log.Errorf("Error creating script: %+v", err)
 		return err
 	}
-	if _, err := createTempScript(tempDir, "gh-ost-on-status", onHookContent(schema.OnlineDDLStatusRunning)); err != nil {
+	if _, err := createTempScript(tempDir, "gh-ost-on-status", onHookContent(schema.OnlineDDLStatusRunning, emptyHint)); err != nil {
 		log.Errorf("Error creating script: %+v", err)
 		return err
 	}
-	if _, err := createTempScript(tempDir, "gh-ost-on-success", onHookContent(schema.OnlineDDLStatusComplete)); err != nil {
+	if _, err := createTempScript(tempDir, "gh-ost-on-success", onHookContent(schema.OnlineDDLStatusComplete, emptyHint)); err != nil {
 		log.Errorf("Error creating script: %+v", err)
 		return err
 	}
-	if _, err := createTempScript(tempDir, "gh-ost-on-failure", onHookContent(schema.OnlineDDLStatusFailed)); err != nil {
+	if _, err := createTempScript(tempDir, "gh-ost-on-failure", onHookContent(schema.OnlineDDLStatusFailed, emptyHint)); err != nil {
+		log.Errorf("Error creating script: %+v", err)
+		return err
+	}
+	if _, err := createTempScript(tempDir, "gh-ost-on-begin-postponed", onHookContent(schema.OnlineDDLStatusRunning, readyToCompleteHint)); err != nil {
 		log.Errorf("Error creating script: %+v", err)
 		return err
 	}
@@ -1296,16 +1302,16 @@ export MYSQL_PWD
 
 	sub before_create_new_table {
 	  my($self, % args) = @_;
-	  get("http://localhost:{{VTTABLET_PORT}}/schema-migration/report-status?uuid={{MIGRATION_UUID}}&status={{OnlineDDLStatusRunning}}&dryrun={{DRYRUN}}");
+	  get("http://localhost:{{VTTABLET_PORT}}/schema-migration/report-status?uuid={{MIGRATION_UUID}}&status={{OnlineDDLStatusRunning}}&hint=&dryrun={{DRYRUN}}");
 	}
 
 	sub before_exit {
 		my($self, % args) = @_;
 		my $exit_status = $args{exit_status};
 	  if ($exit_status == 0) {
-	    get("http://localhost:{{VTTABLET_PORT}}/schema-migration/report-status?uuid={{MIGRATION_UUID}}&status={{OnlineDDLStatusComplete}}&dryrun={{DRYRUN}}");
+	    get("http://localhost:{{VTTABLET_PORT}}/schema-migration/report-status?uuid={{MIGRATION_UUID}}&status={{OnlineDDLStatusComplete}}&hint=&dryrun={{DRYRUN}}");
 	  } else {
-	    get("http://localhost:{{VTTABLET_PORT}}/schema-migration/report-status?uuid={{MIGRATION_UUID}}&status={{OnlineDDLStatusFailed}}&dryrun={{DRYRUN}}");
+	    get("http://localhost:{{VTTABLET_PORT}}/schema-migration/report-status?uuid={{MIGRATION_UUID}}&status={{OnlineDDLStatusFailed}}&hint=&dryrun={{DRYRUN}}");
 	  }
 	}
 
@@ -1635,9 +1641,40 @@ func (e *Executor) scheduleNextMigration(ctx context.Context) error {
 	e.migrationMutex.Lock()
 	defer e.migrationMutex.Unlock()
 
-	// The query sqlScheduleSingleMigration has some business logic; in the future, we can
-	// consider moving the logic outside the query and into this function's code.
-	_, err := e.execQuery(ctx, sqlScheduleSingleMigration)
+	var onlyScheduleOneMigration sync.Once
+
+	r, err := e.execQuery(ctx, sqlSelectQueuedMigrations)
+	if err != nil {
+		return err
+	}
+	for _, row := range r.Named().Rows {
+		uuid := row["migration_uuid"].ToString()
+		postponeCompletion := row.AsBool("postpone_completion", false)
+		readyToComplete := row.AsBool("ready_to_complete", false)
+		ddlAction := row["ddl_action"].ToString()
+
+		if !readyToComplete {
+			// Whether postponsed or not, CREATE and DROP operations are inherently "ready to complete"
+			// because their operation is instantaneous.
+			switch ddlAction {
+			case sqlparser.CreateStr, sqlparser.DropStr:
+				if err := e.updateMigrationReadyToComplete(ctx, uuid, true); err != nil {
+					return err
+				}
+			}
+		}
+		if ddlAction == sqlparser.AlterStr || !postponeCompletion {
+			// Any non-postponed migration can be scheduled
+			// postponed ALTER can be scheduled
+			// We only schedule a single migration in the execution of this function
+			onlyScheduleOneMigration.Do(func() {
+				err = e.updateMigrationStatus(ctx, uuid, schema.OnlineDDLStatusReady)
+			})
+			if err != nil {
+				return err
+			}
+		}
+	}
 	return err
 }
 
@@ -1839,7 +1876,7 @@ func (e *Executor) executeRevert(ctx context.Context, onlineDDL *schema.OnlineDD
 			}
 			if len(artifactTables) == 0 {
 				// This indicates no table was actually created. this must have been a CREATE TABLE IF NOT EXISTS where the table already existed.
-				_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown)
+				_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown, emptyHint)
 			}
 
 			for _, artifactTable := range artifactTables {
@@ -1865,7 +1902,7 @@ func (e *Executor) executeRevert(ctx context.Context, onlineDDL *schema.OnlineDD
 			}
 			if len(artifactTables) == 0 {
 				// Could happen on `DROP TABLE IF EXISTS` where the table did not exist...
-				_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown)
+				_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown, emptyHint)
 			}
 			for _, artifactTable := range artifactTables {
 				if err := e.updateArtifacts(ctx, onlineDDL.UUID, artifactTable); err != nil {
@@ -2282,7 +2319,7 @@ func (e *Executor) executeAlterViewOnline(ctx context.Context, onlineDDL *schema
 	}
 	defer conn.Close()
 
-	_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusRunning, false, progressPctStarted, etaSecondsUnknown, rowsCopiedUnknown)
+	_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusRunning, false, progressPctStarted, etaSecondsUnknown, rowsCopiedUnknown, emptyHint)
 
 	if _, err := conn.ExecuteFetch(artifactViewCreateSQL, 0, false); err != nil {
 		return err
@@ -2309,7 +2346,7 @@ func (e *Executor) executeAlterViewOnline(ctx context.Context, onlineDDL *schema
 		return err
 	}
 
-	_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown)
+	_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown, emptyHint)
 
 	return nil
 }
@@ -2407,7 +2444,7 @@ func (e *Executor) executeMigration(ctx context.Context, onlineDDL *schema.Onlin
 		}
 		if completedUUID != "" {
 			// Yep. We mark this migration as implicitly complete, and we're done with it!
-			_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown)
+			_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown, emptyHint)
 			_ = e.updateMigrationMessage(ctx, onlineDDL.UUID, fmt.Sprintf("duplicate DDL as %s for migration context %s", completedUUID, onlineDDL.MigrationContext))
 			return nil
 		}
@@ -2441,7 +2478,7 @@ func (e *Executor) executeMigration(ctx context.Context, onlineDDL *schema.Onlin
 				// table does exist, so this declarative DROP turns out to really be an actual DROP. No further action is needed here
 			} else {
 				// table does not exist. We mark this DROP as implicitly sucessful
-				_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown)
+				_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown, emptyHint)
 				_ = e.updateMigrationMessage(ctx, onlineDDL.UUID, "no change")
 				return nil
 			}
@@ -2474,7 +2511,7 @@ func (e *Executor) executeMigration(ctx context.Context, onlineDDL *schema.Onlin
 				}
 				if alterClause == "" {
 					// No diff! We mark this CREATE as implicitly sucessful
-					_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown)
+					_ = e.onSchemaMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusComplete, false, progressPctFull, etaSecondsNow, rowsCopiedUnknown, emptyHint)
 					_ = e.updateMigrationMessage(ctx, onlineDDL.UUID, "no change")
 					return nil
 				}
@@ -2831,6 +2868,10 @@ func (e *Executor) reviewRunningMigrations(ctx context.Context) (countRunnning i
 							isReady = false
 						}
 					}
+					// Indicate to outside observers whether the migration is generally ready to complete.
+					// In the case of a postponed migration, we will not complete it, but the user will
+					// understand whether "now is a good time" or "not there yet"
+					_ = e.updateMigrationReadyToComplete(ctx, uuid, isReady)
 					if postponeCompletion {
 						// override. Even if migration is ready, we do not complet it.
 						isReady = false
@@ -3390,6 +3431,18 @@ func (e *Executor) updateMigrationIsView(ctx context.Context, uuid string, isVie
 	return err
 }
 
+func (e *Executor) updateMigrationReadyToComplete(ctx context.Context, uuid string, isReady bool) error {
+	query, err := sqlparser.ParseAndBind(sqlUpdateMigrationReadyToComplete,
+		sqltypes.BoolBindVariable(isReady),
+		sqltypes.StringBindVariable(uuid),
+	)
+	if err != nil {
+		return err
+	}
+	_, err = e.execQuery(ctx, query)
+	return err
+}
+
 // retryMigrationWhere retries a migration based on a given WHERE clause
 func (e *Executor) retryMigrationWhere(ctx context.Context, whereExpr string) (result *sqltypes.Result, err error) {
 	e.migrationMutex.Lock()
@@ -3624,7 +3677,7 @@ func (e *Executor) ShowMigrationLogs(ctx context.Context, stmt *sqlparser.ShowMi
 
 // onSchemaMigrationStatus is called when a status is set/changed for a running migration
 func (e *Executor) onSchemaMigrationStatus(ctx context.Context,
-	uuid string, status schema.OnlineDDLStatus, dryRun bool, progressPct float64, etaSeconds int64, rowsCopied int64) (err error) {
+	uuid string, status schema.OnlineDDLStatus, dryRun bool, progressPct float64, etaSeconds int64, rowsCopied int64, hint string) (err error) {
 	if dryRun && status != schema.OnlineDDLStatusFailed {
 		// We don't consider dry-run reports unless there's a failure
 		return nil
@@ -3666,6 +3719,11 @@ func (e *Executor) onSchemaMigrationStatus(ctx context.Context,
 	if err := e.updateRowsCopied(ctx, uuid, rowsCopied); err != nil {
 		return err
 	}
+	if hint == readyToCompleteHint {
+		if err := e.updateMigrationReadyToComplete(ctx, uuid, true); err != nil {
+			return err
+		}
+	}
 	if !dryRun {
 		switch status {
 		case schema.OnlineDDLStatusComplete, schema.OnlineDDLStatusFailed:
@@ -3678,7 +3736,7 @@ func (e *Executor) onSchemaMigrationStatus(ctx context.Context,
 
 // OnSchemaMigrationStatus is called by TabletServer's API, which is invoked by a running gh-ost migration's hooks.
 func (e *Executor) OnSchemaMigrationStatus(ctx context.Context,
-	uuidParam, statusParam, dryrunParam, progressParam, etaParam, rowsCopiedParam string) (err error) {
+	uuidParam, statusParam, dryrunParam, progressParam, etaParam, rowsCopiedParam, hint string) (err error) {
 	status := schema.OnlineDDLStatus(statusParam)
 	dryRun := (dryrunParam == "true")
 	var progressPct float64
@@ -3694,7 +3752,7 @@ func (e *Executor) OnSchemaMigrationStatus(ctx context.Context,
 		rowsCopied = rows
 	}
 
-	return e.onSchemaMigrationStatus(ctx, uuidParam, status, dryRun, progressPct, etaSeconds, rowsCopied)
+	return e.onSchemaMigrationStatus(ctx, uuidParam, status, dryRun, progressPct, etaSeconds, rowsCopied, hint)
 }
 
 // VExec is called by a VExec invocation
diff --git a/go/vt/vttablet/onlineddl/schema.go b/go/vt/vttablet/onlineddl/schema.go
index 736c091bf1..f9f3980477 100644
--- a/go/vt/vttablet/onlineddl/schema.go
+++ b/go/vt/vttablet/onlineddl/schema.go
@@ -72,6 +72,7 @@ const (
 	alterSchemaMigrationsTableRevertedUUID             = "ALTER TABLE _vt.schema_migrations add column reverted_uuid varchar(64) NOT NULL DEFAULT ''"
 	alterSchemaMigrationsTableRevertedUUIDIndex        = "ALTER TABLE _vt.schema_migrations add KEY reverted_uuid_idx (reverted_uuid(64))"
 	alterSchemaMigrationsTableIsView                   = "ALTER TABLE _vt.schema_migrations add column is_view tinyint unsigned NOT NULL DEFAULT 0"
+	alterSchemaMigrationsTableReadyToComplete          = "ALTER TABLE _vt.schema_migrations add column ready_to_complete tinyint unsigned NOT NULL DEFAULT 0"
 
 	sqlInsertMigration = `INSERT IGNORE INTO _vt.schema_migrations (
 		migration_uuid,
@@ -96,19 +97,15 @@ const (
 		%a, %a, %a, %a, %a, %a, %a, %a, %a, FROM_UNIXTIME(NOW()), %a, %a, %a, %a, %a, %a, %a, %a
 	)`
 
-	sqlScheduleSingleMigration = `UPDATE _vt.schema_migrations
-		SET
-			migration_status='ready',
-			ready_timestamp=NOW()
+	sqlSelectQueuedMigrations = `SELECT
+			migration_uuid,
+			ddl_action,
+			postpone_completion,
+			ready_to_complete
+		FROM _vt.schema_migrations
 		WHERE
 			migration_status='queued'
-			AND (
-				postpone_completion=0 OR ddl_action='alter'
-			)
-		ORDER BY
-			requested_timestamp ASC
-		LIMIT 1
-	`  // if the migration is CREATE or DROP, and postpone_completion=1, we just don't schedule it
+	`
 	sqlUpdateMySQLTable = `UPDATE _vt.schema_migrations
 			SET mysql_table=%a
 		WHERE
@@ -138,7 +135,12 @@ const (
 			SET is_view=%a
 		WHERE
 			migration_uuid=%a
-`
+	`
+	sqlUpdateMigrationReadyToComplete = `UPDATE _vt.schema_migrations
+			SET ready_to_complete=%a
+		WHERE
+			migration_uuid=%a
+	`
 	sqlUpdateMigrationStartedTimestamp = `UPDATE _vt.schema_migrations SET
 			started_timestamp =IFNULL(started_timestamp,  NOW()),
 			liveness_timestamp=IFNULL(liveness_timestamp, NOW())
@@ -367,6 +369,7 @@ const (
 			migration_context,
 			retain_artifacts_seconds,
 			is_view,
+			ready_to_complete,
 			postpone_completion
 		FROM _vt.schema_migrations
 		WHERE
@@ -571,4 +574,5 @@ var ApplyDDL = []string{
 	alterSchemaMigrationsTableRevertedUUID,
 	alterSchemaMigrationsTableRevertedUUIDIndex,
 	alterSchemaMigrationsTableIsView,
+	alterSchemaMigrationsTableReadyToComplete,
 }
diff --git a/go/vt/vttablet/tabletserver/tabletserver.go b/go/vt/vttablet/tabletserver/tabletserver.go
index 69454efc42..7090805877 100644
--- a/go/vt/vttablet/tabletserver/tabletserver.go
+++ b/go/vt/vttablet/tabletserver/tabletserver.go
@@ -1619,7 +1619,7 @@ func (tsv *TabletServer) registerMigrationStatusHandler() {
 	tsv.exporter.HandleFunc("/schema-migration/report-status", func(w http.ResponseWriter, r *http.Request) {
 		ctx := tabletenv.LocalContext()
 		query := r.URL.Query()
-		if err := tsv.onlineDDLExecutor.OnSchemaMigrationStatus(ctx, query.Get("uuid"), query.Get("status"), query.Get("dryrun"), query.Get("progress"), query.Get("eta"), query.Get("rowscopied")); err != nil {
+		if err := tsv.onlineDDLExecutor.OnSchemaMigrationStatus(ctx, query.Get("uuid"), query.Get("status"), query.Get("dryrun"), query.Get("progress"), query.Get("eta"), query.Get("rowscopied"), query.Get("hint")); err != nil {
 			http.Error(w, fmt.Sprintf("not ok: %v", err), http.StatusInternalServerError)
 			return
 		}


Test patch:
diff --git a/go/test/endtoend/onlineddl/ghost/onlineddl_ghost_test.go b/go/test/endtoend/onlineddl/ghost/onlineddl_ghost_test.go
index a1c6200b07..cb512b92e7 100644
--- a/go/test/endtoend/onlineddl/ghost/onlineddl_ghost_test.go
+++ b/go/test/endtoend/onlineddl/ghost/onlineddl_ghost_test.go
@@ -175,53 +175,10 @@ func TestMain(m *testing.M) {
 			"-ddl_strategy", "gh-ost",
 		}
 
-		primaryTablet := *clusterInstance.Keyspaces[0].Shards[0].Vttablets[0]
-		replicaTablet := *clusterInstance.Keyspaces[0].Shards[0].Vttablets[1]
-
-		tableACLConfigJSON := path.Join("/tmp", "table_acl_config.json")
-		f, _ := os.Create(tableACLConfigJSON)
-		vtStaticACLJSON := `{
-	"table_groups":
-	[
-	    {
-	        "name": "vitess user groups",
-	        "table_names_or_prefixes":
-	        [
-	            "%"
-	        ],
-	        "readers":
-	        [
-	            "vitess-reader",
-	            "vitess-writer",
-	            "vitess-admin"
-	        ],
-	        "writers":
-	        [
-	            "vitess-writer",
-	            "vitess-writer-only",
-	            "vitess-admin"
-	        ],
-	        "admins":
-	        [
-	            "vitess-admin"
-	        ]
-	    }
-	]
-}`
-		f.WriteString(vtStaticACLJSON)
-		f.Close()
-
-		// start the tablets
-		for _, tablet := range []cluster.Vttablet{primaryTablet, replicaTablet} {
-			tablet.VttabletProcess.ExtraArgs = append(tablet.VttabletProcess.ExtraArgs, "-table-acl-config", tableACLConfigJSON, "-queryserver-config-strict-table-acl")
-			tablet.VttabletProcess.Setup()
-		}
-
 		if err := clusterInstance.StartTopo(); err != nil {
 			return 1, err
 		}
 
-		// Start keyspace
 		keyspace := &cluster.Keyspace{
 			Name:    keyspaceName,
 			VSchema: vSchema,
@@ -425,7 +382,7 @@ func testWithInitialSchema(t *testing.T) {
 }
 
 // testOnlineDDLStatement runs an online DDL, ALTER statement
-func testOnlineDDLStatement(t *testing.T, alterStatement string, ddlStrategy string, executeStrategy string, expectHint string, callerId string) (uuid string) {
+func testOnlineDDLStatement(t *testing.T, alterStatement string, ddlStrategy string, executeStrategy string, expectHint string, callerID string) (uuid string) {
 	tableName := fmt.Sprintf("vt_onlineddl_test_%02d", 3)
 	sqlQuery := fmt.Sprintf(alterStatement, tableName)
 	if executeStrategy == "vtgate" {
@@ -435,7 +392,7 @@ func testOnlineDDLStatement(t *testing.T, alterStatement string, ddlStrategy str
 		}
 	} else {
 		var err error
-		uuid, err = clusterInstance.VtctlclientProcess.ApplySchemaWithOutput(keyspaceName, sqlQuery, cluster.VtctlClientParams{DDLStrategy: ddlStrategy, CallerId: callerId})
+		uuid, err = clusterInstance.VtctlclientProcess.ApplySchemaWithOutput(keyspaceName, sqlQuery, cluster.VtctlClientParams{DDLStrategy: ddlStrategy, CallerId: callerID})
 		assert.NoError(t, err)
 	}
 	uuid = strings.TrimSpace(uuid)
diff --git a/go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go b/go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go
index da6c48b6a5..955b8f04bb 100644
--- a/go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go
+++ b/go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go
@@ -303,6 +303,14 @@ func TestSchemaChange(t *testing.T) {
 			onlineddl.WaitForMigrationStatus(t, &vtParams, shards, t1uuid, 20*time.Second, schema.OnlineDDLStatusRunning)
 			onlineddl.WaitForMigrationStatus(t, &vtParams, shards, t2uuid, 20*time.Second, schema.OnlineDDLStatusRunning)
 		})
+		t.Run("test ready-to-complete", func(t *testing.T) {
+			rs := onlineddl.ReadMigrations(t, &vtParams, t1uuid)
+			require.NotNil(t, rs)
+			for _, row := range rs.Named().Rows {
+				readyToComplete := row.AsInt64("ready_to_complete", 0)
+				assert.Equal(t, int64(1), readyToComplete)
+			}
+		})
 		t.Run("complete t2", func(t *testing.T) {
 			// now that both are running, let's unblock t2. We expect it to complete.
 			// Issue a complete and wait for successful completion
@@ -465,6 +473,14 @@ func TestSchemaChange(t *testing.T) {
 			// drop1 migration should block. It can run concurrently to t1, but conflicts on table name
 			onlineddl.CheckMigrationStatus(t, &vtParams, shards, drop1uuid, schema.OnlineDDLStatusReady)
 		})
+		t.Run("t3 ready to complete", func(t *testing.T) {
+			rs := onlineddl.ReadMigrations(t, &vtParams, drop1uuid)
+			require.NotNil(t, rs)
+			for _, row := range rs.Named().Rows {
+				readyToComplete := row.AsInt64("ready_to_complete", 0)
+				assert.Equal(t, int64(1), readyToComplete)
+			}
+		})
 		t.Run("t3drop complete", func(t *testing.T) {
 			// drop3 migration should not block. It can run concurrently to t1, and does not conflict
 			// even though t1drop is blocked! This is the heart of this test
diff --git a/go/test/endtoend/onlineddl/vrepl/onlineddl_vrepl_test.go b/go/test/endtoend/onlineddl/vrepl/onlineddl_vrepl_test.go
index a6f75f6c0a..02695565d7 100644
--- a/go/test/endtoend/onlineddl/vrepl/onlineddl_vrepl_test.go
+++ b/go/test/endtoend/onlineddl/vrepl/onlineddl_vrepl_test.go
@@ -528,6 +528,14 @@ func TestSchemaChange(t *testing.T) {
 	}
 	t.Run("Online DROP, vtctl", func(t *testing.T) {
 		uuid := testOnlineDDLStatement(t, onlineDDLDropTableStatement, "online", providedUUID, providedMigrationContext, "vtctl", "", "", false)
+		t.Run("test ready to complete", func(t *testing.T) {
+			rs := onlineddl.ReadMigrations(t, &vtParams, uuid)
+			require.NotNil(t, rs)
+			for _, row := range rs.Named().Rows {
+				readyToComplete := row.AsInt64("ready_to_complete", 0)
+				assert.Equal(t, int64(1), readyToComplete)
+			}
+		})
 		onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusComplete)
 		onlineddl.CheckCancelMigration(t, &vtParams, shards, uuid, false)
 		onlineddl.CheckRetryMigration(t, &vtParams, shards, uuid, false)
diff --git a/go/test/endtoend/reparent/newfeaturetest/reparent_test.go b/go/test/endtoend/reparent/newfeaturetest/reparent_test.go
index 8f8a228145..df88479414 100644
--- a/go/test/endtoend/reparent/newfeaturetest/reparent_test.go
+++ b/go/test/endtoend/reparent/newfeaturetest/reparent_test.go
@@ -274,3 +274,35 @@ func TestRecoverWithMultipleFailures(t *testing.T) {
 	newPrimary := utils.GetNewPrimary(t, clusterInstance)
 	utils.ConfirmReplication(t, newPrimary, []*cluster.Vttablet{tablets[2], tablets[3]})
 }
+
+// TestERSFailFast tests that ERS will fail fast if it cannot find any tablet which can be safely promoted instead of promoting
+// a tablet and hanging while inserting a row in the reparent journal on getting semi-sync ACKs
+func TestERSFailFast(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
+	defer utils.TeardownCluster(clusterInstance)
+	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
+	utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{tablets[1], tablets[2], tablets[3]})
+
+	// make tablets[1] a rdonly tablet.
+	err := clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", tablets[1].Alias, "rdonly")
+	require.NoError(t, err)
+
+	// Confirm that replication is still working as intended
+	utils.ConfirmReplication(t, tablets[0], tablets[1:])
+
+	strChan := make(chan string)
+	go func() {
+		// We expect this to fail since we have ignored all replica tablets and only the rdonly is left, which is not capable of sending semi-sync ACKs
+		out, err := utils.ErsIgnoreTablet(clusterInstance, tablets[2], "240s", "90s", []*cluster.Vttablet{tablets[0], tablets[3]}, false)
+		require.Error(t, err)
+		strChan <- out
+	}()
+
+	select {
+	case out := <-strChan:
+		require.Contains(t, out, "proposed primary zone1-0000000103 will not be able to make forward progress on being promoted")
+	case <-time.After(60 * time.Second):
+		require.Fail(t, "Emergency Reparent Shard did not fail in 60 seconds")
+	}
+}
diff --git a/go/vt/vtctl/reparentutil/durability_funcs_test.go b/go/vt/vtctl/reparentutil/durability_funcs_test.go
index e9fa1fc592..01578a4071 100644
--- a/go/vt/vtctl/reparentutil/durability_funcs_test.go
+++ b/go/vt/vtctl/reparentutil/durability_funcs_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package reparentutil
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -313,3 +314,70 @@ func Test_haveRevoked(t *testing.T) {
 		})
 	}
 }
+
+func Test_canEstablishForTablet(t *testing.T) {
+	tests := []struct {
+		name             string
+		durabilityPolicy string
+		primaryEligible  *topodatapb.Tablet
+		tabletsReached   []*topodatapb.Tablet
+		canEstablish     bool
+	}{
+		{
+			name:             "primary not reached",
+			durabilityPolicy: "none",
+			primaryEligible:  primaryTablet,
+			tabletsReached: []*topodatapb.Tablet{
+				replicaTablet, replicaCrossCellTablet, rdonlyCrossCellTablet, rdonlyTablet,
+			},
+			canEstablish: false,
+		}, {
+			name:             "not established",
+			durabilityPolicy: "semi_sync",
+			primaryEligible:  primaryTablet,
+			tabletsReached: []*topodatapb.Tablet{
+				primaryTablet, rdonlyCrossCellTablet, rdonlyTablet,
+			},
+			canEstablish: false,
+		}, {
+			name:             "not established",
+			durabilityPolicy: "cross_cell",
+			primaryEligible:  primaryTablet,
+			tabletsReached: []*topodatapb.Tablet{
+				primaryTablet, replicaTablet, rdonlyCrossCellTablet, rdonlyTablet,
+			},
+			canEstablish: false,
+		}, {
+			name:             "established",
+			durabilityPolicy: "none",
+			primaryEligible:  primaryTablet,
+			tabletsReached: []*topodatapb.Tablet{
+				primaryTablet,
+			},
+			canEstablish: true,
+		}, {
+			name:             "established",
+			durabilityPolicy: "semi_sync",
+			primaryEligible:  primaryTablet,
+			tabletsReached: []*topodatapb.Tablet{
+				primaryTablet, replicaTablet,
+			},
+			canEstablish: true,
+		}, {
+			name:             "established",
+			durabilityPolicy: "cross_cell",
+			primaryEligible:  primaryTablet,
+			tabletsReached: []*topodatapb.Tablet{
+				primaryTablet, replicaCrossCellTablet,
+			},
+			canEstablish: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(fmt.Sprintf("'%s' durability policy - %s", tt.durabilityPolicy, tt.name), func(t *testing.T) {
+			err := SetDurabilityPolicy(tt.durabilityPolicy)
+			require.NoError(t, err)
+			require.Equalf(t, tt.canEstablish, canEstablishForTablet(tt.primaryEligible, tt.tabletsReached), "canEstablishForTablet(%v, %v)", tt.primaryEligible, tt.tabletsReached)
+		})
+	}
+}
diff --git a/go/vt/vtctl/reparentutil/emergency_reparenter_test.go b/go/vt/vtctl/reparentutil/emergency_reparenter_test.go
index 2680b91391..dd2f829b53 100644
--- a/go/vt/vtctl/reparentutil/emergency_reparenter_test.go
+++ b/go/vt/vtctl/reparentutil/emergency_reparenter_test.go
@@ -1279,7 +1279,7 @@ func TestEmergencyReparenter_reparentShardLocked(t *testing.T) {
 			errShouldContain: "failed to be upgraded to primary",
 		},
 		{
-			name:                 "constraint failure - promotion-rule",
+			name:                 "promotion-rule - no valid candidates for emergency reparent",
 			durability:           "none",
 			emergencyReparentOps: EmergencyReparentOptions{},
 			tmc: &testutil.TabletManagerClient{
@@ -1392,10 +1392,131 @@ func TestEmergencyReparenter_reparentShardLocked(t *testing.T) {
 			shard:            "-",
 			ts:               memorytopo.NewServer("zone1"),
 			shouldErr:        true,
-			errShouldContain: "elected primary does not satisfy promotion rule constraint",
+			errShouldContain: "no valid candidates for emergency reparent",
+		},
+		{
+			name:       "proposed primary - must not promotion rule",
+			durability: "none",
+			emergencyReparentOps: EmergencyReparentOptions{
+				NewPrimaryAlias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  102,
+				},
+			},
+			tmc: &testutil.TabletManagerClient{
+				PopulateReparentJournalResults: map[string]error{
+					"zone1-0000000102": nil,
+				},
+				PromoteReplicaResults: map[string]struct {
+					Result string
+					Error  error
+				}{
+					"zone1-0000000102": {
+						Result: "ok",
+						Error:  nil,
+					},
+				},
+				PrimaryPositionResults: map[string]struct {
+					Position string
+					Error    error
+				}{
+					"zone1-0000000102": {
+						Error: nil,
+					},
+				},
+				SetReplicationSourceResults: map[string]error{
+					"zone1-0000000100": nil,
+					"zone1-0000000101": nil,
+				},
+				StopReplicationAndGetStatusResults: map[string]struct {
+					Status     *replicationdatapb.Status
+					StopStatus *replicationdatapb.StopReplicationStatus
+					Error      error
+				}{
+					"zone1-0000000100": {
+						StopStatus: &replicationdatapb.StopReplicationStatus{
+							Before: &replicationdatapb.Status{},
+							After: &replicationdatapb.Status{
+								SourceUuid:       "3E11FA47-71CA-11E1-9E33-C80AA9429562",
+								RelayLogPosition: "MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-21",
+							},
+						},
+					},
+					"zone1-0000000101": {
+						StopStatus: &replicationdatapb.StopReplicationStatus{
+							Before: &replicationdatapb.Status{},
+							After: &replicationdatapb.Status{
+								SourceUuid:       "3E11FA47-71CA-11E1-9E33-C80AA9429562",
+								RelayLogPosition: "MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-21",
+							},
+						},
+					},
+					"zone1-0000000102": {
+						StopStatus: &replicationdatapb.StopReplicationStatus{
+							Before: &replicationdatapb.Status{},
+							After: &replicationdatapb.Status{
+								SourceUuid:       "3E11FA47-71CA-11E1-9E33-C80AA9429562",
+								RelayLogPosition: "MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-26",
+							},
+						},
+					},
+				},
+				WaitForPositionResults: map[string]map[string]error{
+					"zone1-0000000100": {
+						"MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-21": nil,
+					},
+					"zone1-0000000101": {
+						"MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-21": nil,
+					},
+					"zone1-0000000102": {
+						"MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-26": nil,
+					},
+				},
+			},
+			shards: []*vtctldatapb.Shard{
+				{
+					Keyspace: "testkeyspace",
+					Name:     "-",
+				},
+			},
+			tablets: []*topodatapb.Tablet{
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  100,
+					},
+					Type:     topodatapb.TabletType_RDONLY,
+					Keyspace: "testkeyspace",
+					Shard:    "-",
+				},
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  101,
+					},
+					Type:     topodatapb.TabletType_RDONLY,
+					Keyspace: "testkeyspace",
+					Shard:    "-",
+				},
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  102,
+					},
+					Type:     topodatapb.TabletType_RDONLY,
+					Keyspace: "testkeyspace",
+					Shard:    "-",
+					Hostname: "proposed primary",
+				},
+			},
+			keyspace:         "testkeyspace",
+			shard:            "-",
+			ts:               memorytopo.NewServer("zone1"),
+			shouldErr:        true,
+			errShouldContain: "proposed primary zone1-0000000102 has a must not promotion rule",
 		},
 		{
-			name:                 "constraint failure - cross-cell",
+			name:                 "cross cell - no valid candidates",
 			durability:           "none",
 			emergencyReparentOps: EmergencyReparentOptions{PreventCrossCellPromotion: true},
 			tmc: &testutil.TabletManagerClient{
@@ -1520,7 +1641,265 @@ func TestEmergencyReparenter_reparentShardLocked(t *testing.T) {
 			shard:            "-",
 			ts:               memorytopo.NewServer("zone1", "zone2"),
 			shouldErr:        true,
-			errShouldContain: "elected primary does not satisfy geographic constraint",
+			errShouldContain: "no valid candidates for emergency reparent",
+		},
+		{
+			name:       "proposed primary in a different cell",
+			durability: "none",
+			emergencyReparentOps: EmergencyReparentOptions{
+				PreventCrossCellPromotion: true,
+				NewPrimaryAlias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  102,
+				},
+			},
+			tmc: &testutil.TabletManagerClient{
+				PopulateReparentJournalResults: map[string]error{
+					"zone1-0000000102": nil,
+				},
+				PromoteReplicaResults: map[string]struct {
+					Result string
+					Error  error
+				}{
+					"zone1-0000000102": {
+						Result: "ok",
+						Error:  nil,
+					},
+				},
+				PrimaryPositionResults: map[string]struct {
+					Position string
+					Error    error
+				}{
+					"zone1-0000000102": {
+						Error: nil,
+					},
+				},
+				SetReplicationSourceResults: map[string]error{
+					"zone1-0000000100": nil,
+					"zone1-0000000101": nil,
+				},
+				StopReplicationAndGetStatusResults: map[string]struct {
+					Status     *replicationdatapb.Status
+					StopStatus *replicationdatapb.StopReplicationStatus
+					Error      error
+				}{
+					"zone1-0000000100": {
+						StopStatus: &replicationdatapb.StopReplicationStatus{
+							Before: &replicationdatapb.Status{},
+							After: &replicationdatapb.Status{
+								SourceUuid:       "3E11FA47-71CA-11E1-9E33-C80AA9429562",
+								RelayLogPosition: "MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-21",
+							},
+						},
+					},
+					"zone1-0000000101": {
+						StopStatus: &replicationdatapb.StopReplicationStatus{
+							Before: &replicationdatapb.Status{},
+							After: &replicationdatapb.Status{
+								SourceUuid:       "3E11FA47-71CA-11E1-9E33-C80AA9429562",
+								RelayLogPosition: "MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-21",
+							},
+						},
+					},
+					"zone1-0000000102": {
+						StopStatus: &replicationdatapb.StopReplicationStatus{
+							Before: &replicationdatapb.Status{},
+							After: &replicationdatapb.Status{
+								SourceUuid:       "3E11FA47-71CA-11E1-9E33-C80AA9429562",
+								RelayLogPosition: "MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-26",
+							},
+						},
+					},
+				},
+				WaitForPositionResults: map[string]map[string]error{
+					"zone1-0000000100": {
+						"MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-21": nil,
+					},
+					"zone1-0000000101": {
+						"MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-21": nil,
+					},
+					"zone1-0000000102": {
+						"MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-26": nil,
+					},
+				},
+			},
+			shards: []*vtctldatapb.Shard{
+				{
+					Keyspace: "testkeyspace",
+					Name:     "-",
+					Shard: &topodatapb.Shard{
+						PrimaryAlias: &topodatapb.TabletAlias{
+							Cell: "zone2",
+							Uid:  100,
+						},
+					},
+				},
+			},
+			tablets: []*topodatapb.Tablet{
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  100,
+					},
+					Keyspace: "testkeyspace",
+					Shard:    "-",
+				},
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  101,
+					},
+					Keyspace: "testkeyspace",
+					Shard:    "-",
+				},
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  102,
+					},
+					Keyspace: "testkeyspace",
+					Shard:    "-",
+					Hostname: "proposed primary",
+				},
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone2",
+						Uid:  100,
+					},
+					Keyspace: "testkeyspace",
+					Shard:    "-",
+					Hostname: "failed previous primary",
+				},
+			},
+			keyspace:         "testkeyspace",
+			shard:            "-",
+			ts:               memorytopo.NewServer("zone1", "zone2"),
+			shouldErr:        true,
+			errShouldContain: "proposed primary zone1-0000000102 is is a different cell as the previous primary",
+		},
+		{
+			name:       "proposed primary cannot make progress",
+			durability: "cross_cell",
+			emergencyReparentOps: EmergencyReparentOptions{
+				NewPrimaryAlias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  102,
+				},
+			},
+			tmc: &testutil.TabletManagerClient{
+				PopulateReparentJournalResults: map[string]error{
+					"zone1-0000000102": nil,
+				},
+				PromoteReplicaResults: map[string]struct {
+					Result string
+					Error  error
+				}{
+					"zone1-0000000102": {
+						Result: "ok",
+						Error:  nil,
+					},
+				},
+				PrimaryPositionResults: map[string]struct {
+					Position string
+					Error    error
+				}{
+					"zone1-0000000102": {
+						Error: nil,
+					},
+				},
+				SetReplicationSourceResults: map[string]error{
+					"zone1-0000000100": nil,
+					"zone1-0000000101": nil,
+				},
+				StopReplicationAndGetStatusResults: map[string]struct {
+					Status     *replicationdatapb.Status
+					StopStatus *replicationdatapb.StopReplicationStatus
+					Error      error
+				}{
+					"zone1-0000000100": {
+						StopStatus: &replicationdatapb.StopReplicationStatus{
+							Before: &replicationdatapb.Status{},
+							After: &replicationdatapb.Status{
+								SourceUuid:       "3E11FA47-71CA-11E1-9E33-C80AA9429562",
+								RelayLogPosition: "MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-21",
+							},
+						},
+					},
+					"zone1-0000000101": {
+						StopStatus: &replicationdatapb.StopReplicationStatus{
+							Before: &replicationdatapb.Status{},
+							After: &replicationdatapb.Status{
+								SourceUuid:       "3E11FA47-71CA-11E1-9E33-C80AA9429562",
+								RelayLogPosition: "MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-21",
+							},
+						},
+					},
+					"zone1-0000000102": {
+						StopStatus: &replicationdatapb.StopReplicationStatus{
+							Before: &replicationdatapb.Status{},
+							After: &replicationdatapb.Status{
+								SourceUuid:       "3E11FA47-71CA-11E1-9E33-C80AA9429562",
+								RelayLogPosition: "MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-26",
+							},
+						},
+					},
+				},
+				WaitForPositionResults: map[string]map[string]error{
+					"zone1-0000000100": {
+						"MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-21": nil,
+					},
+					"zone1-0000000101": {
+						"MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-21": nil,
+					},
+					"zone1-0000000102": {
+						"MySQL56/3E11FA47-71CA-11E1-9E33-C80AA9429562:1-26": nil,
+					},
+				},
+			},
+			shards: []*vtctldatapb.Shard{
+				{
+					Keyspace: "testkeyspace",
+					Name:     "-",
+					Shard: &topodatapb.Shard{
+						PrimaryAlias: &topodatapb.TabletAlias{
+							Cell: "zone1",
+							Uid:  101,
+						},
+					},
+				},
+			},
+			tablets: []*topodatapb.Tablet{
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  100,
+					},
+					Keyspace: "testkeyspace",
+					Shard:    "-",
+				},
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  101,
+					},
+					Keyspace: "testkeyspace",
+					Shard:    "-",
+				},
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  102,
+					},
+					Keyspace: "testkeyspace",
+					Shard:    "-",
+					Hostname: "proposed primary",
+				},
+			},
+			keyspace:         "testkeyspace",
+			shard:            "-",
+			ts:               memorytopo.NewServer("zone1"),
+			shouldErr:        true,
+			errShouldContain: "proposed primary zone1-0000000102 will not be able to make forward progress on being promoted",
 		},
 	}
 
@@ -2729,130 +3108,42 @@ func TestEmergencyReparenter_findMostAdvanced(t *testing.T) {
 						},
 					},
 				},
-				"zone1-0000000102": {
-					Tablet: &topodatapb.Tablet{
-						Alias: &topodatapb.TabletAlias{
-							Cell: "zone1",
-							Uid:  102,
-						},
-						Type: topodatapb.TabletType_RDONLY,
-					},
-				},
-				"zone1-0000000404": {
-					Tablet: &topodatapb.Tablet{
-						Alias: &topodatapb.TabletAlias{
-							Cell: "zone1",
-							Uid:  404,
-						},
-						Hostname: "ignored tablet",
-					},
-				},
-			},
-			err: "split brain detected between servers",
-		},
-	}
-
-	_ = SetDurabilityPolicy("none")
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			erp := NewEmergencyReparenter(nil, nil, logutil.NewMemoryLogger())
-
-			winningTablet, _, err := erp.findMostAdvanced(test.validCandidates, test.tabletMap, test.emergencyReparentOps)
-			if test.err != "" {
-				assert.Error(t, err)
-				assert.Contains(t, err.Error(), test.err)
-			} else {
-				assert.NoError(t, err)
-				assert.True(t, topoproto.TabletAliasEqual(test.result.Alias, winningTablet.Alias))
-			}
-		})
-	}
-}
-
-func TestEmergencyReparenter_checkIfConstraintsSatisfied(t *testing.T) {
-	testcases := []struct {
-		name                    string
-		newPrimary, prevPrimary *topodatapb.Tablet
-		opts                    EmergencyReparentOptions
-		err                     string
-	}{
-		{
-			name: "no constraint failure",
-			newPrimary: &topodatapb.Tablet{
-				Alias: &topodatapb.TabletAlias{
-					Cell: "cell1",
-				},
-				Type: topodatapb.TabletType_REPLICA,
-			},
-			prevPrimary: &topodatapb.Tablet{
-				Alias: &topodatapb.TabletAlias{
-					Cell: "cell1",
-				},
-			},
-			opts: EmergencyReparentOptions{PreventCrossCellPromotion: true},
-			err:  "",
-		}, {
-			name: "promotion rule constraint failure",
-			newPrimary: &topodatapb.Tablet{
-				Alias: &topodatapb.TabletAlias{
-					Cell: "cell1",
-					Uid:  100,
-				},
-				Type: topodatapb.TabletType_RDONLY,
-			},
-			prevPrimary: &topodatapb.Tablet{
-				Alias: &topodatapb.TabletAlias{
-					Cell: "cell1",
-				},
-			},
-			opts: EmergencyReparentOptions{PreventCrossCellPromotion: true},
-			err:  "elected primary does not satisfy promotion rule constraint - cell1-0000000100",
-		}, {
-			name: "cross cell constraint failure",
-			newPrimary: &topodatapb.Tablet{
-				Alias: &topodatapb.TabletAlias{
-					Cell: "cell1",
-					Uid:  100,
-				},
-				Type: topodatapb.TabletType_REPLICA,
-			},
-			prevPrimary: &topodatapb.Tablet{
-				Alias: &topodatapb.TabletAlias{
-					Cell: "cell2",
-				},
-			},
-			opts: EmergencyReparentOptions{PreventCrossCellPromotion: true},
-			err:  "elected primary does not satisfy geographic constraint - cell1-0000000100",
-		}, {
-			name: "cross cell but no constraint failure",
-			newPrimary: &topodatapb.Tablet{
-				Alias: &topodatapb.TabletAlias{
-					Cell: "cell1",
-					Uid:  100,
-				},
-				Type: topodatapb.TabletType_REPLICA,
-			},
-			prevPrimary: &topodatapb.Tablet{
-				Alias: &topodatapb.TabletAlias{
-					Cell: "cell2",
+				"zone1-0000000102": {
+					Tablet: &topodatapb.Tablet{
+						Alias: &topodatapb.TabletAlias{
+							Cell: "zone1",
+							Uid:  102,
+						},
+						Type: topodatapb.TabletType_RDONLY,
+					},
+				},
+				"zone1-0000000404": {
+					Tablet: &topodatapb.Tablet{
+						Alias: &topodatapb.TabletAlias{
+							Cell: "zone1",
+							Uid:  404,
+						},
+						Hostname: "ignored tablet",
+					},
 				},
 			},
-			opts: EmergencyReparentOptions{PreventCrossCellPromotion: false},
-			err:  "",
+			err: "split brain detected between servers",
 		},
 	}
 
 	_ = SetDurabilityPolicy("none")
-	erp := NewEmergencyReparenter(nil, nil, nil)
 
-	for _, testcase := range testcases {
-		t.Run(testcase.name, func(t *testing.T) {
-			err := erp.checkIfConstraintsSatisfied(testcase.newPrimary, testcase.prevPrimary, testcase.opts)
-			if testcase.err == "" {
-				require.NoError(t, err)
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			erp := NewEmergencyReparenter(nil, nil, logutil.NewMemoryLogger())
+
+			winningTablet, _, err := erp.findMostAdvanced(test.validCandidates, test.tabletMap, test.emergencyReparentOps)
+			if test.err != "" {
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), test.err)
 			} else {
-				require.EqualError(t, err, testcase.err)
+				assert.NoError(t, err)
+				assert.True(t, topoproto.TabletAliasEqual(test.result.Alias, winningTablet.Alias))
 			}
 		})
 	}
@@ -3253,20 +3544,21 @@ func TestEmergencyReparenter_promoteIntermediateSource(t *testing.T) {
 	t.Parallel()
 
 	tests := []struct {
-		name                 string
-		emergencyReparentOps EmergencyReparentOptions
-		tmc                  *testutil.TabletManagerClient
-		unlockTopo           bool
-		newSourceTabletAlias string
-		ts                   *topo.Server
-		keyspace             string
-		shard                string
-		tablets              []*topodatapb.Tablet
-		tabletMap            map[string]*topo.TabletInfo
-		statusMap            map[string]*replicationdatapb.StopReplicationStatus
-		shouldErr            bool
-		errShouldContain     string
-		result               []*topodatapb.Tablet
+		name                  string
+		emergencyReparentOps  EmergencyReparentOptions
+		tmc                   *testutil.TabletManagerClient
+		unlockTopo            bool
+		newSourceTabletAlias  string
+		ts                    *topo.Server
+		keyspace              string
+		shard                 string
+		tablets               []*topodatapb.Tablet
+		validCandidateTablets []*topodatapb.Tablet
+		tabletMap             map[string]*topo.TabletInfo
+		statusMap             map[string]*replicationdatapb.StopReplicationStatus
+		shouldErr             bool
+		errShouldContain      string
+		result                []*topodatapb.Tablet
 	}{
 		{
 			name:                 "success",
@@ -3366,6 +3658,27 @@ func TestEmergencyReparenter_promoteIntermediateSource(t *testing.T) {
 					Hostname: "requires force start",
 				},
 			},
+			validCandidateTablets: []*topodatapb.Tablet{
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  100,
+					},
+					Hostname: "primary-elect",
+				}, {
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  101,
+					},
+				},
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  102,
+					},
+					Hostname: "requires force start",
+				},
+			},
 		},
 		{
 			name:                 "all replicas failed",
@@ -3416,6 +3729,27 @@ func TestEmergencyReparenter_promoteIntermediateSource(t *testing.T) {
 					},
 				},
 			},
+			validCandidateTablets: []*topodatapb.Tablet{
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  100,
+					},
+					Hostname: "primary-elect",
+				}, {
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  101,
+					},
+				},
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  102,
+					},
+					Hostname: "requires force start",
+				},
+			},
 			statusMap:        map[string]*replicationdatapb.StopReplicationStatus{},
 			keyspace:         "testkeyspace",
 			shard:            "-",
@@ -3475,12 +3809,129 @@ func TestEmergencyReparenter_promoteIntermediateSource(t *testing.T) {
 			shard:     "-",
 			ts:        memorytopo.NewServer("zone1"),
 			shouldErr: false,
+			validCandidateTablets: []*topodatapb.Tablet{
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  100,
+					},
+					Hostname: "primary-elect",
+				}, {
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  101,
+					},
+				},
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  102,
+					},
+					Hostname: "requires force start",
+				},
+			},
+			result: []*topodatapb.Tablet{
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  100,
+					},
+				}, {
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  101,
+					},
+				},
+			},
+		},
+		{
+			name:                 "success - filter using valid candidate list",
+			emergencyReparentOps: EmergencyReparentOptions{},
+			tmc: &testutil.TabletManagerClient{
+				PopulateReparentJournalResults: map[string]error{
+					"zone1-0000000100": nil,
+				},
+				PrimaryPositionResults: map[string]struct {
+					Position string
+					Error    error
+				}{
+					"zone1-0000000100": {
+						Error: nil,
+					},
+				},
+				SetReplicationSourceResults: map[string]error{
+					"zone1-0000000101": nil,
+					"zone1-0000000102": nil,
+				},
+			},
+			newSourceTabletAlias: "zone1-0000000100",
+			tabletMap: map[string]*topo.TabletInfo{
+				"zone1-0000000100": {
+					Tablet: &topodatapb.Tablet{
+						Alias: &topodatapb.TabletAlias{
+							Cell: "zone1",
+							Uid:  100,
+						},
+						Hostname: "primary-elect",
+					},
+				},
+				"zone1-0000000101": {
+					Tablet: &topodatapb.Tablet{
+						Alias: &topodatapb.TabletAlias{
+							Cell: "zone1",
+							Uid:  101,
+						},
+					},
+				},
+				"zone1-0000000102": {
+					Tablet: &topodatapb.Tablet{
+						Alias: &topodatapb.TabletAlias{
+							Cell: "zone1",
+							Uid:  102,
+						},
+						Hostname: "requires force start",
+					},
+				},
+			},
+			statusMap: map[string]*replicationdatapb.StopReplicationStatus{
+				"zone1-0000000101": { // forceStart = false
+					Before: &replicationdatapb.Status{
+						IoThreadRunning:  false,
+						SqlThreadRunning: false,
+					},
+				},
+				"zone1-0000000102": { // forceStart = true
+					Before: &replicationdatapb.Status{
+						IoThreadRunning:  true,
+						SqlThreadRunning: true,
+					},
+				},
+			},
+			keyspace:  "testkeyspace",
+			shard:     "-",
+			ts:        memorytopo.NewServer("zone1"),
+			shouldErr: false,
 			result: []*topodatapb.Tablet{
 				{
 					Alias: &topodatapb.TabletAlias{
 						Cell: "zone1",
 						Uid:  100,
 					},
+					Hostname: "primary-elect",
+				}, {
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  101,
+					},
+				},
+			},
+			validCandidateTablets: []*topodatapb.Tablet{
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  100,
+					},
+					Hostname: "primary-elect",
 				}, {
 					Alias: &topodatapb.TabletAlias{
 						Cell: "zone1",
@@ -3491,6 +3942,7 @@ func TestEmergencyReparenter_promoteIntermediateSource(t *testing.T) {
 		},
 	}
 
+	_ = SetDurabilityPolicy("none")
 	for _, tt := range tests {
 		tt := tt
 
@@ -3523,7 +3975,7 @@ func TestEmergencyReparenter_promoteIntermediateSource(t *testing.T) {
 			tabletInfo := tt.tabletMap[tt.newSourceTabletAlias]
 
 			erp := NewEmergencyReparenter(tt.ts, tt.tmc, logger)
-			res, err := erp.promoteIntermediateSource(ctx, ev, tabletInfo.Tablet, tt.tabletMap, tt.statusMap, tt.emergencyReparentOps)
+			res, err := erp.promoteIntermediateSource(ctx, ev, tabletInfo.Tablet, tt.tabletMap, tt.statusMap, tt.validCandidateTablets, tt.emergencyReparentOps)
 			if tt.shouldErr {
 				assert.Error(t, err)
 				assert.Contains(t, err.Error(), tt.errShouldContain)
@@ -3531,7 +3983,10 @@ func TestEmergencyReparenter_promoteIntermediateSource(t *testing.T) {
 			}
 
 			assert.NoError(t, err)
-			assert.ElementsMatch(t, tt.result, res)
+			require.Equal(t, len(tt.result), len(res))
+			for idx, tablet := range res {
+				assert.EqualValues(t, topoproto.TabletAliasString(tt.result[idx].Alias), topoproto.TabletAliasString(tablet.Alias))
+			}
 		})
 	}
 }
@@ -3579,6 +4034,10 @@ func TestEmergencyReparenter_identifyPrimaryCandidate(t *testing.T) {
 					Uid:  100,
 				},
 			},
+		}, {
+			name:            "empty valid list",
+			validCandidates: nil,
+			err:             "no valid candidates for emergency reparent",
 		}, {
 			name: "explicit request for a primary tablet not in valid list",
 			emergencyReparentOps: EmergencyReparentOptions{NewPrimaryAlias: &topodatapb.TabletAlias{
@@ -3587,7 +4046,14 @@ func TestEmergencyReparenter_identifyPrimaryCandidate(t *testing.T) {
 			}},
 			intermediateSource: nil,
 			prevPrimary:        nil,
-			validCandidates:    nil,
+			validCandidates: []*topodatapb.Tablet{
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  101,
+					},
+				},
+			},
 			tabletMap: map[string]*topo.TabletInfo{
 				"zone1-0000000100": {
 					Tablet: &topodatapb.Tablet{
@@ -3982,3 +4448,141 @@ func TestParentContextCancelled(t *testing.T) {
 	_, err := erp.reparentReplicas(ctx, ev, tabletMap[newPrimaryTabletAlias].Tablet, tabletMap, statusMap, emergencyReparentOps, false, false)
 	require.NoError(t, err)
 }
+
+func TestEmergencyReparenter_filterValidCandidates(t *testing.T) {
+	var (
+		primaryTablet = &topodatapb.Tablet{
+			Alias: &topodatapb.TabletAlias{
+				Cell: "zone-1",
+				Uid:  1,
+			},
+			Type: topodatapb.TabletType_PRIMARY,
+		}
+		replicaTablet = &topodatapb.Tablet{
+			Alias: &topodatapb.TabletAlias{
+				Cell: "zone-1",
+				Uid:  2,
+			},
+			Type: topodatapb.TabletType_REPLICA,
+		}
+		rdonlyTablet = &topodatapb.Tablet{
+			Alias: &topodatapb.TabletAlias{
+				Cell: "zone-1",
+				Uid:  3,
+			},
+			Type: topodatapb.TabletType_RDONLY,
+		}
+		replicaCrossCellTablet = &topodatapb.Tablet{
+			Alias: &topodatapb.TabletAlias{
+				Cell: "zone-2",
+				Uid:  2,
+			},
+			Type: topodatapb.TabletType_REPLICA,
+		}
+		rdonlyCrossCellTablet = &topodatapb.Tablet{
+			Alias: &topodatapb.TabletAlias{
+				Cell: "zone-2",
+				Uid:  3,
+			},
+			Type: topodatapb.TabletType_RDONLY,
+		}
+	)
+	allTablets := []*topodatapb.Tablet{primaryTablet, replicaTablet, rdonlyTablet, replicaCrossCellTablet, rdonlyCrossCellTablet}
+	tests := []struct {
+		name             string
+		durability       string
+		validTablets     []*topodatapb.Tablet
+		tabletsReachable []*topodatapb.Tablet
+		prevPrimary      *topodatapb.Tablet
+		opts             EmergencyReparentOptions
+		filteredTablets  []*topodatapb.Tablet
+		errShouldContain string
+	}{
+		{
+			name:             "filter must not",
+			durability:       "none",
+			validTablets:     allTablets,
+			tabletsReachable: allTablets,
+			filteredTablets:  []*topodatapb.Tablet{primaryTablet, replicaTablet, replicaCrossCellTablet},
+		}, {
+			name:             "filter cross cell",
+			durability:       "none",
+			validTablets:     allTablets,
+			tabletsReachable: allTablets,
+			prevPrimary: &topodatapb.Tablet{
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone-1",
+				},
+			},
+			opts: EmergencyReparentOptions{
+				PreventCrossCellPromotion: true,
+			},
+			filteredTablets: []*topodatapb.Tablet{primaryTablet, replicaTablet},
+		}, {
+			name:             "filter establish",
+			durability:       "cross_cell",
+			validTablets:     []*topodatapb.Tablet{primaryTablet, replicaTablet},
+			tabletsReachable: []*topodatapb.Tablet{primaryTablet, replicaTablet, rdonlyTablet, rdonlyCrossCellTablet},
+			filteredTablets:  nil,
+		}, {
+			name:       "filter mixed",
+			durability: "cross_cell",
+			prevPrimary: &topodatapb.Tablet{
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone-2",
+				},
+			},
+			opts: EmergencyReparentOptions{
+				PreventCrossCellPromotion: true,
+			},
+			validTablets:     allTablets,
+			tabletsReachable: allTablets,
+			filteredTablets:  []*topodatapb.Tablet{replicaCrossCellTablet},
+		}, {
+			name:             "error - requested primary must not",
+			durability:       "none",
+			validTablets:     allTablets,
+			tabletsReachable: allTablets,
+			opts: EmergencyReparentOptions{
+				NewPrimaryAlias: rdonlyTablet.Alias,
+			},
+			errShouldContain: "proposed primary zone-1-0000000003 has a must not promotion rule",
+		}, {
+			name:             "error - requested primary not in same cell",
+			durability:       "none",
+			validTablets:     allTablets,
+			tabletsReachable: allTablets,
+			prevPrimary:      primaryTablet,
+			opts: EmergencyReparentOptions{
+				PreventCrossCellPromotion: true,
+				NewPrimaryAlias:           replicaCrossCellTablet.Alias,
+			},
+			errShouldContain: "proposed primary zone-2-0000000002 is is a different cell as the previous primary",
+		}, {
+			name:             "error - requested primary cannot establish",
+			durability:       "cross_cell",
+			validTablets:     allTablets,
+			tabletsReachable: []*topodatapb.Tablet{primaryTablet, replicaTablet, rdonlyTablet, rdonlyCrossCellTablet},
+			opts: EmergencyReparentOptions{
+				NewPrimaryAlias: primaryTablet.Alias,
+			},
+			errShouldContain: "proposed primary zone-1-0000000001 will not be able to make forward progress on being promoted",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := SetDurabilityPolicy(tt.durability)
+			require.NoError(t, err)
+			logger := logutil.NewMemoryLogger()
+			erp := NewEmergencyReparenter(nil, nil, logger)
+			tabletList, err := erp.filterValidCandidates(tt.validTablets, tt.tabletsReachable, tt.prevPrimary, tt.opts)
+			if tt.errShouldContain != "" {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), tt.errShouldContain)
+			} else {
+				require.NoError(t, err)
+				require.EqualValues(t, tt.filteredTablets, tabletList)
+			}
+		})
+	}
+}
diff --git a/go/vt/vtctl/reparentutil/replication_test.go b/go/vt/vtctl/reparentutil/replication_test.go
index b3a3973657..d50c048776 100644
--- a/go/vt/vtctl/reparentutil/replication_test.go
+++ b/go/vt/vtctl/reparentutil/replication_test.go
@@ -209,7 +209,7 @@ func TestFindValidEmergencyReparentCandidates(t *testing.T) {
 
 // stopReplicationAndBuildStatusMapsTestTMClient implements
 // tmclient.TabletManagerClient to facilitate testing of
-// StopReplicationAndBuildStatusMaps.
+// stopReplicationAndBuildStatusMaps.
 type stopReplicationAndBuildStatusMapsTestTMClient struct {
 	tmclient.TabletManagerClient
 
@@ -270,7 +270,7 @@ func (fake *stopReplicationAndBuildStatusMapsTestTMClient) StopReplicationAndGet
 	return nil, nil, assert.AnError
 }
 
-func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
+func Test_stopReplicationAndBuildStatusMaps(t *testing.T) {
 	t.Parallel()
 
 	ctx := context.Background()
@@ -285,6 +285,7 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 		tabletToWaitFor          *topodatapb.TabletAlias
 		expectedStatusMap        map[string]*replicationdatapb.StopReplicationStatus
 		expectedPrimaryStatusMap map[string]*replicationdatapb.PrimaryStatus
+		expectedTabletsReachable []*topodatapb.Tablet
 		shouldErr                bool
 	}{
 		{
@@ -341,7 +342,20 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 				},
 			},
 			expectedPrimaryStatusMap: map[string]*replicationdatapb.PrimaryStatus{},
-			shouldErr:                false,
+			expectedTabletsReachable: []*topodatapb.Tablet{{
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  100,
+				},
+			}, {
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  101,
+				},
+			}},
+			shouldErr: false,
 		},
 		{
 			name:       "success - 2 rdonly failures",
@@ -421,7 +435,20 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 				},
 			},
 			expectedPrimaryStatusMap: map[string]*replicationdatapb.PrimaryStatus{},
-			shouldErr:                false,
+			expectedTabletsReachable: []*topodatapb.Tablet{{
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  100,
+				},
+			}, {
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  101,
+				},
+			}},
+			shouldErr: false,
 		},
 		{
 			name:       "success - 1 rdonly and 1 replica failures",
@@ -501,7 +528,20 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 				},
 			},
 			expectedPrimaryStatusMap: map[string]*replicationdatapb.PrimaryStatus{},
-			shouldErr:                false,
+			expectedTabletsReachable: []*topodatapb.Tablet{{
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  100,
+				},
+			}, {
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  101,
+				},
+			}},
+			shouldErr: false,
 		},
 		{
 			name:       "ignore tablets",
@@ -552,6 +592,13 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 					After:  &replicationdatapb.Status{Position: "101-after"},
 				},
 			},
+			expectedTabletsReachable: []*topodatapb.Tablet{{
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  101,
+				},
+			}},
 			expectedPrimaryStatusMap: map[string]*replicationdatapb.PrimaryStatus{},
 			shouldErr:                false,
 		},
@@ -618,6 +665,19 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 					Position: "primary-position-100",
 				},
 			},
+			expectedTabletsReachable: []*topodatapb.Tablet{{
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  100,
+				},
+			}, {
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  101,
+				},
+			}},
 			shouldErr: false,
 		},
 		{
@@ -675,7 +735,14 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 				},
 			},
 			expectedPrimaryStatusMap: map[string]*replicationdatapb.PrimaryStatus{}, // zone1-0000000100 fails to demote, so does not appear
-			shouldErr:                false,
+			expectedTabletsReachable: []*topodatapb.Tablet{{
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  101,
+				},
+			}},
+			shouldErr: false,
 		},
 		{
 			name:       "multiple tablets are PRIMARY and cannot demote",
@@ -727,6 +794,7 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 			ignoredTablets:           sets.NewString(),
 			expectedStatusMap:        nil,
 			expectedPrimaryStatusMap: nil,
+			expectedTabletsReachable: nil,
 			shouldErr:                true, // we get multiple errors, so we fail
 		},
 		{
@@ -782,6 +850,13 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 					After:  &replicationdatapb.Status{Position: "101-after"},
 				},
 			},
+			expectedTabletsReachable: []*topodatapb.Tablet{{
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  101,
+				},
+			}},
 			expectedPrimaryStatusMap: map[string]*replicationdatapb.PrimaryStatus{},
 			shouldErr:                false,
 		},
@@ -832,7 +907,14 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 				},
 			},
 			expectedPrimaryStatusMap: map[string]*replicationdatapb.PrimaryStatus{},
-			shouldErr:                false,
+			expectedTabletsReachable: []*topodatapb.Tablet{{
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  101,
+				},
+			}},
+			shouldErr: false,
 		},
 		{
 			name:       "multiple tablets fail StopReplication",
@@ -873,6 +955,7 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 			ignoredTablets:           sets.NewString(),
 			expectedStatusMap:        nil,
 			expectedPrimaryStatusMap: nil,
+			expectedTabletsReachable: nil,
 			shouldErr:                true,
 		},
 		{
@@ -954,6 +1037,25 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 					After:  &replicationdatapb.Status{Position: "102-after"},
 				},
 			},
+			expectedTabletsReachable: []*topodatapb.Tablet{{
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  100,
+				},
+			}, {
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  101,
+				},
+			}, {
+				Type: topodatapb.TabletType_REPLICA,
+				Alias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  102,
+				},
+			}},
 			waitReplicasTimeout:      time.Minute,
 			expectedPrimaryStatusMap: map[string]*replicationdatapb.PrimaryStatus{},
 			shouldErr:                false,
@@ -966,15 +1068,19 @@ func TestStopReplicationAndBuildStatusMaps(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			err := SetDurabilityPolicy(tt.durability)
 			require.NoError(t, err)
-			statusMap, primaryStatusMap, err := StopReplicationAndBuildStatusMaps(ctx, tt.tmc, &events.Reparent{}, tt.tabletMap, tt.waitReplicasTimeout, tt.ignoredTablets, tt.tabletToWaitFor, logger)
+			res, err := stopReplicationAndBuildStatusMaps(ctx, tt.tmc, &events.Reparent{}, tt.tabletMap, tt.waitReplicasTimeout, tt.ignoredTablets, tt.tabletToWaitFor, logger)
 			if tt.shouldErr {
 				assert.Error(t, err)
 				return
 			}
 
 			assert.NoError(t, err)
-			assert.Equal(t, tt.expectedStatusMap, statusMap, "StopReplicationStatus mismatch")
-			assert.Equal(t, tt.expectedPrimaryStatusMap, primaryStatusMap, "PrimaryStatusMap mismatch")
+			assert.Equal(t, tt.expectedStatusMap, res.statusMap, "StopReplicationStatus mismatch")
+			assert.Equal(t, tt.expectedPrimaryStatusMap, res.primaryStatusMap, "PrimaryStatusMap mismatch")
+			require.Equal(t, len(tt.expectedTabletsReachable), len(res.reachableTablets), "TabletsReached length mismatch")
+			for idx, tablet := range res.reachableTablets {
+				assert.True(t, topoproto.IsTabletInList(tablet, tt.expectedTabletsReachable), "TabletsReached[%d] not found - %s", idx, topoproto.TabletAliasString(tablet.Alias))
+			}
 		})
 	}
 }
diff --git a/go/vt/vtgate/engine/ordered_aggregate_test.go b/go/vt/vtgate/engine/ordered_aggregate_test.go
index 3de0ca0f9e..e8fcc0cac4 100644
--- a/go/vt/vtgate/engine/ordered_aggregate_test.go
+++ b/go/vt/vtgate/engine/ordered_aggregate_test.go
@@ -656,94 +656,17 @@ func TestMerge(t *testing.T) {
 		"1|3|2.8|2|bc",
 	)
 
-	merged, _, err := oa.merge(fields, r.Rows[0], r.Rows[1], nil, nil)
+	merged, _, err := merge(fields, r.Rows[0], r.Rows[1], nil, nil, oa.Aggregates)
 	assert.NoError(err)
 	want := sqltypes.MakeTestResult(fields, "1|5|6.0|2|bc").Rows[0]
 	assert.Equal(want, merged)
 
 	// swap and retry
-	merged, _, err = oa.merge(fields, r.Rows[1], r.Rows[0], nil, nil)
+	merged, _, err = merge(fields, r.Rows[1], r.Rows[0], nil, nil, oa.Aggregates)
 	assert.NoError(err)
 	assert.Equal(want, merged)
 }
 
-func TestNoInputAndNoGroupingKeys(outer *testing.T) {
-	testCases := []struct {
-		name        string
-		opcode      AggregateOpcode
-		expectedVal string
-		expectedTyp string
-	}{{
-		"count(distinct col1)",
-		AggregateCountDistinct,
-		"0",
-		"int64",
-	}, {
-		"col1",
-		AggregateCount,
-		"0",
-		"int64",
-	}, {
-		"sum(distinct col1)",
-		AggregateSumDistinct,
-		"null",
-		"decimal",
-	}, {
-		"col1",
-		AggregateSum,
-		"null",
-		"int64",
-	}, {
-		"col1",
-		AggregateMax,
-		"null",
-		"int64",
-	}, {
-		"col1",
-		AggregateMin,
-		"null",
-		"int64",
-	}}
-
-	for _, test := range testCases {
-		outer.Run(test.name, func(t *testing.T) {
-			assert := assert.New(t)
-			fp := &fakePrimitive{
-				results: []*sqltypes.Result{sqltypes.MakeTestResult(
-					sqltypes.MakeTestFields(
-						"col1",
-						"int64",
-					),
-					// Empty input table
-				)},
-			}
-
-			oa := &OrderedAggregate{
-				PreProcess: true,
-				Aggregates: []*AggregateParams{{
-					Opcode: test.opcode,
-					Col:    0,
-					Alias:  test.name,
-				}},
-				GroupByKeys: []*GroupByParams{},
-				Input:       fp,
-			}
-
-			result, err := oa.TryExecute(&noopVCursor{}, nil, false)
-			assert.NoError(err)
-
-			wantResult := sqltypes.MakeTestResult(
-				sqltypes.MakeTestFields(
-					test.name,
-					test.expectedTyp,
-				),
-				test.expectedVal,
-			)
-			assert.Equal(wantResult, result)
-		})
-	}
-}
-
 func TestOrderedAggregateExecuteGtid(t *testing.T) {
 	vgtid := binlogdatapb.VGtid{}
 	vgtid.ShardGtids = append(vgtid.ShardGtids, &binlogdatapb.ShardGtid{
@@ -756,7 +679,6 @@ func TestOrderedAggregateExecuteGtid(t *testing.T) {
 		Shard:    "80-",
 		Gtid:     "b",
 	})
-	fmt.Println(vgtid.String())
 
 	fp := &fakePrimitive{
 		results: []*sqltypes.Result{sqltypes.MakeTestResult(
@@ -1079,7 +1001,7 @@ func TestOrderedAggregateCollate(t *testing.T) {
 	collationID, _ := collations.Local().LookupID("utf8mb4_0900_ai_ci")
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    1,
 		}},
 		GroupByKeys: []*GroupByParams{{KeyCol: 0, CollationID: collationID}},
@@ -1122,7 +1044,7 @@ func TestOrderedAggregateCollateAS(t *testing.T) {
 	collationID, _ := collations.Local().LookupID("utf8mb4_0900_as_ci")
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    1,
 		}},
 		GroupByKeys: []*GroupByParams{{KeyCol: 0, CollationID: collationID}},
@@ -1167,7 +1089,7 @@ func TestOrderedAggregateCollateKS(t *testing.T) {
 	collationID, _ := collations.Local().LookupID("utf8mb4_ja_0900_as_cs_ks")
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateCount,
+			Opcode: AggregateSum,
 			Col:    1,
 		}},
 		GroupByKeys: []*GroupByParams{{KeyCol: 0, CollationID: collationID}},
diff --git a/go/vt/vtgate/engine/scalar_aggregation_test.go b/go/vt/vtgate/engine/scalar_aggregation_test.go
new file mode 100644
index 0000000000..0e8987e4d4
--- /dev/null
+++ b/go/vt/vtgate/engine/scalar_aggregation_test.go
@@ -0,0 +1,135 @@
+/*
+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 engine
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/sqltypes"
+)
+
+func TestEmptyRows(outer *testing.T) {
+	testCases := []struct {
+		opcode      AggregateOpcode
+		expectedVal string
+		expectedTyp string
+	}{{
+		opcode:      AggregateCountDistinct,
+		expectedVal: "0",
+		expectedTyp: "int64",
+	}, {
+		opcode:      AggregateCount,
+		expectedVal: "0",
+		expectedTyp: "int64",
+	}, {
+		opcode:      AggregateSumDistinct,
+		expectedVal: "null",
+		expectedTyp: "decimal",
+	}, {
+		opcode:      AggregateSum,
+		expectedVal: "null",
+		expectedTyp: "int64",
+	}, {
+		opcode:      AggregateMax,
+		expectedVal: "null",
+		expectedTyp: "int64",
+	}, {
+		opcode:      AggregateMin,
+		expectedVal: "null",
+		expectedTyp: "int64",
+	}}
+
+	for _, test := range testCases {
+		outer.Run(test.opcode.String(), func(t *testing.T) {
+			assert := assert.New(t)
+			fp := &fakePrimitive{
+				results: []*sqltypes.Result{sqltypes.MakeTestResult(
+					sqltypes.MakeTestFields(
+						test.opcode.String(),
+						"int64",
+					),
+					// Empty input table
+				)},
+			}
+
+			oa := &ScalarAggregate{
+				PreProcess: true,
+				Aggregates: []*AggregateParams{{
+					Opcode: test.opcode,
+					Col:    0,
+					Alias:  test.opcode.String(),
+				}},
+				Input: fp,
+			}
+
+			result, err := oa.TryExecute(&noopVCursor{}, nil, false)
+			assert.NoError(err)
+
+			wantResult := sqltypes.MakeTestResult(
+				sqltypes.MakeTestFields(
+					test.opcode.String(),
+					test.expectedTyp,
+				),
+				test.expectedVal,
+			)
+			assert.Equal(wantResult, result)
+		})
+	}
+}
+
+func TestScalarAggregateStreamExecute(t *testing.T) {
+	assert := assert.New(t)
+	fields := sqltypes.MakeTestFields(
+		"count(*)",
+		"uint64",
+	)
+	fp := &fakePrimitive{
+		allResultsInOneCall: true,
+		results: []*sqltypes.Result{
+			sqltypes.MakeTestResult(fields,
+				"1",
+			), sqltypes.MakeTestResult(fields,
+				"3",
+			)},
+	}
+
+	oa := &ScalarAggregate{
+		Aggregates: []*AggregateParams{{
+			Opcode: AggregateCount,
+			Col:    0,
+		}},
+		Input:               fp,
+		TruncateColumnCount: 1,
+		PreProcess:          true,
+	}
+
+	var results []*sqltypes.Result
+	err := oa.TryStreamExecute(&noopVCursor{}, nil, true, func(qr *sqltypes.Result) error {
+		results = append(results, qr)
+		return nil
+	})
+	assert.NoError(err)
+	// one for the fields, and one for the actual aggregation result
+	require.EqualValues(t, 2, len(results), "number of results")
+
+	got := fmt.Sprintf("%v", results[1].Rows)
+	assert.Equal("[[UINT64(4)]]", got)
+}
diff --git a/go/vt/vtgate/evalengine/comparisons_test.go b/go/vt/vtgate/evalengine/comparisons_test.go
index e53b42b911..2db80b196c 100644
--- a/go/vt/vtgate/evalengine/comparisons_test.go
+++ b/go/vt/vtgate/evalengine/comparisons_test.go
@@ -59,9 +59,14 @@ func (tc testCase) run(t *testing.T) {
 	if tc.bv == nil {
 		tc.bv = map[string]*querypb.BindVariable{}
 	}
+	fields := make([]*querypb.Field, len(tc.row))
+	for i, value := range tc.row {
+		fields[i] = &querypb.Field{Type: value.Type()}
+	}
 	env := &ExpressionEnv{
 		BindVars: tc.bv,
 		Row:      tc.row,
+		Fields:   fields,
 	}
 	cmp, err := translateComparisonExpr2(tc.op, tc.v1, tc.v2)
 	if err != nil {
diff --git a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
index 2677992171..797dea2683 100644
--- a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
@@ -280,7 +280,7 @@ Gen4 plan same as above
   "Original": "select count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0)",
     "Inputs": [
       {
@@ -302,7 +302,7 @@ Gen4 plan same as above
   "Original": "select count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0) AS count(*)",
     "Inputs": [
       {
@@ -327,7 +327,7 @@ Gen4 plan same as above
   "Original": "select sum(col) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "sum(0)",
     "Inputs": [
       {
@@ -349,7 +349,7 @@ Gen4 plan same as above
   "Original": "select sum(col) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "sum(0) AS sum(col)",
     "Inputs": [
       {
@@ -374,7 +374,7 @@ Gen4 plan same as above
   "Original": "select min(col) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "min(0)",
     "Inputs": [
       {
@@ -396,7 +396,7 @@ Gen4 plan same as above
   "Original": "select min(col) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "min(0) AS min(col)",
     "Inputs": [
       {
@@ -421,7 +421,7 @@ Gen4 plan same as above
   "Original": "select max(col) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "max(0)",
     "Inputs": [
       {
@@ -443,7 +443,7 @@ Gen4 plan same as above
   "Original": "select max(col) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "max(0) AS max(col)",
     "Inputs": [
       {
@@ -870,7 +870,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0)",
         "Inputs": [
           {
@@ -900,7 +900,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -927,7 +927,7 @@ Gen4 plan same as above
   "Original": "select id, count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(1)",
     "Inputs": [
       {
@@ -1130,7 +1130,7 @@ Gen4 plan same as above
   "Original": "select count(distinct col2) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count_distinct(0) AS count(distinct col2)",
     "Inputs": [
       {
@@ -1154,7 +1154,7 @@ Gen4 plan same as above
   "Original": "select count(distinct col2) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count_distinct(0|1) AS count(distinct col2)",
     "ResultColumns": 1,
     "Inputs": [
@@ -1625,7 +1625,7 @@ Gen4 plan same as above
   "Original": "select count(*) from user order by null",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0)",
     "Inputs": [
       {
@@ -1647,7 +1647,7 @@ Gen4 plan same as above
   "Original": "select count(*) from user order by null",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0) AS count(*)",
     "Inputs": [
       {
@@ -2133,7 +2133,7 @@ Gen4 plan same as above
   "Original": "select count(*) from (select user.col, user_extra.extra from user join user_extra on user.id = user_extra.user_id order by user_extra.extra) a",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0) AS count(*)",
     "Inputs": [
       {
@@ -2214,7 +2214,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0)",
         "Inputs": [
           {
@@ -2238,7 +2238,7 @@ Gen4 plan same as above
   "Original": "select distinct count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0) AS count(*)",
     "Inputs": [
       {
@@ -2311,7 +2311,7 @@ Gen4 plan same as above
   "Original": "select count(distinct a), count(distinct b) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count_distinct(0|2) AS count(distinct a), count_distinct(1|3) AS count(distinct b)",
     "ResultColumns": 2,
     "Inputs": [
@@ -2472,7 +2472,7 @@ Gen4 plan same as above
   "Original": "select count(*) from user where exists (select 1 from user_extra where user_id = user.id group by user_id having max(col) \u003e 10)",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0)",
     "Inputs": [
       {
@@ -2494,7 +2494,7 @@ Gen4 plan same as above
   "Original": "select count(*) from user where exists (select 1 from user_extra where user_id = user.id group by user_id having max(col) \u003e 10)",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0) AS count(*)",
     "Inputs": [
       {
@@ -2743,7 +2743,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2775,7 +2775,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2807,7 +2807,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2839,7 +2839,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2871,7 +2871,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2903,7 +2903,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2935,7 +2935,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2967,7 +2967,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
diff --git a/go/vt/vtgate/planbuilder/testdata/from_cases.txt b/go/vt/vtgate/planbuilder/testdata/from_cases.txt
index a2e917d0b1..03fc842fce 100644
--- a/go/vt/vtgate/planbuilder/testdata/from_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/from_cases.txt
@@ -3700,7 +3700,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0)",
         "Inputs": [
           {
@@ -3730,7 +3730,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
diff --git a/go/vt/vtgate/planbuilder/testdata/oltp_cases.txt b/go/vt/vtgate/planbuilder/testdata/oltp_cases.txt
index 94a973c370..c8563f514b 100644
--- a/go/vt/vtgate/planbuilder/testdata/oltp_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/oltp_cases.txt
@@ -47,7 +47,7 @@ Gen4 plan same as above
   "Original": "SELECT SUM(k) FROM sbtest43 WHERE id BETWEEN 90 AND 990",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "sum(0)",
     "Inputs": [
       {
@@ -69,7 +69,7 @@ Gen4 plan same as above
   "Original": "SELECT SUM(k) FROM sbtest43 WHERE id BETWEEN 90 AND 990",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "sum(0) AS SUM(k)",
     "Inputs": [
       {
diff --git a/go/vt/vtgate/planbuilder/testdata/postprocess_cases.txt b/go/vt/vtgate/planbuilder/testdata/postprocess_cases.txt
index 88abec9aae..003543cf68 100644
--- a/go/vt/vtgate/planbuilder/testdata/postprocess_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/postprocess_cases.txt
@@ -1802,7 +1802,7 @@ Gen4 plan same as above
   "Original": "select count(id), num from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0)",
     "Inputs": [
       {
@@ -1834,7 +1834,7 @@ Gen4 error: In aggregated query without GROUP BY, expression of SELECT list cont
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0)",
         "ResultColumns": 3,
         "Inputs": [
@@ -2236,7 +2236,7 @@ Gen4 error: Expression of SELECT list is not in GROUP BY clause and contains non
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
diff --git a/go/vt/vtgate/planbuilder/testdata/select_cases.txt b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
index 17f921e3fc..55ce84405e 100644
--- a/go/vt/vtgate/planbuilder/testdata/select_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
@@ -82,7 +82,7 @@ Gen4 plan same as above
   "Original": "select /*vt+ QUERY_TIMEOUT_MS=1000 */ count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0)",
     "Inputs": [
       {
@@ -105,7 +105,7 @@ Gen4 plan same as above
   "Original": "select /*vt+ QUERY_TIMEOUT_MS=1000 */ count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0) AS count(*)",
     "Inputs": [
       {
@@ -177,7 +177,7 @@ Gen4 plan same as above
   "Original": "select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0)",
     "Inputs": [
       {
@@ -200,7 +200,7 @@ Gen4 plan same as above
   "Original": "select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0) AS count(*)",
     "Inputs": [
       {
@@ -226,7 +226,7 @@ Gen4 plan same as above
   "Original": "/*VT_SPAN_CONTEXT=123*/select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0)",
     "Inputs": [
       {
@@ -249,7 +249,7 @@ Gen4 plan same as above
   "Original": "/*VT_SPAN_CONTEXT=123*/select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(0) AS count(*)",
     "Inputs": [
       {
@@ -1766,7 +1766,7 @@ Gen4 plan same as above
       },
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0)",
         "Inputs": [
           {
@@ -1810,7 +1810,7 @@ Gen4 plan same as above
       },
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS count(*)",
         "Inputs": [
           {
@@ -1902,7 +1902,7 @@ Gen4 plan same as above
       },
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0)",
         "Inputs": [
           {
@@ -1948,7 +1948,7 @@ Gen4 plan same as above
       },
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "count(0) AS count(*)",
         "Inputs": [
           {
diff --git a/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt b/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt
index 8266e3e553..2c8971bb99 100644
--- a/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt
@@ -30,7 +30,7 @@ Gen4 plan same as above
   "Original": "select sum(l_extendedprice * l_discount) as revenue from lineitem where l_shipdate \u003e= date('1994-01-01') and l_shipdate \u003c date('1994-01-01') + interval '1' year and l_discount between 0.06 - 0.01 and 0.06 + 0.01 and l_quantity \u003c 24",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "sum(0)",
     "Inputs": [
       {
@@ -52,7 +52,7 @@ Gen4 plan same as above
   "Original": "select sum(l_extendedprice * l_discount) as revenue from lineitem where l_shipdate \u003e= date('1994-01-01') and l_shipdate \u003c date('1994-01-01') + interval '1' year and l_discount between 0.06 - 0.01 and 0.06 + 0.01 and l_quantity \u003c 24",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "sum(0) AS revenue",
     "Inputs": [
       {
@@ -129,7 +129,7 @@ Gen4 error: unsupported: in scatter query: complex aggregate expression
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "max(0)",
         "Inputs": [
           {
@@ -200,7 +200,7 @@ Gen4 error: unsupported: in scatter query: complex aggregate expression
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Ordered",
+        "Variant": "Scalar",
         "Aggregates": "max(0) AS max(total_revenue)",
         "Inputs": [
           {
@@ -438,7 +438,7 @@ Gen4 error: unsupported: cross-shard correlated subquery
   "Original": "select sum(l_extendedprice* (1 - l_discount)) as revenue from lineitem, part where ( p_partkey = l_partkey and p_brand = 'Brand#12' and p_container in ('SM CASE', 'SM BOX', 'SM PACK', 'SM PKG') and l_quantity \u003e= 1 and l_quantity \u003c= 1 + 10 and p_size between 1 and 5 and l_shipmode in ('AIR', 'AIR REG') and l_shipinstruct = 'DELIVER IN PERSON' ) or ( p_partkey = l_partkey and p_brand = 'Brand#23' and p_container in ('MED BAG', 'MED BOX', 'MED PKG', 'MED PACK') and l_quantity \u003e= 10 and l_quantity \u003c= 10 + 10 and p_size between 1 and 10 and l_shipmode in ('AIR', 'AIR REG') and l_shipinstruct = 'DELIVER IN PERSON' ) or ( p_partkey = l_partkey and p_brand = 'Brand#34' and p_container in ('LG CASE', 'LG BOX', 'LG PACK', 'LG PKG') and l_quantity \u003e= 20 and l_quantity \u003c= 20 + 10 and p_size between 1 and 15 and l_shipmode in ('AIR', 'AIR REG') and l_shipinstruct = 'DELIVER IN PERSON' )",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "sum(0) AS revenue",
     "Inputs": [
       {
diff --git a/go/vt/vtgate/planbuilder/testdata/union_cases.txt b/go/vt/vtgate/planbuilder/testdata/union_cases.txt
index 3ab3ff0e67..57d8c6f504 100644
--- a/go/vt/vtgate/planbuilder/testdata/union_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/union_cases.txt
@@ -1241,7 +1241,7 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
         "Inputs": [
           {
             "OperatorType": "Aggregate",
-            "Variant": "Ordered",
+            "Variant": "Scalar",
             "Aggregates": "count(0)",
             "Inputs": [
               {
@@ -1259,7 +1259,7 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
           },
           {
             "OperatorType": "Aggregate",
-            "Variant": "Ordered",
+            "Variant": "Scalar",
             "Aggregates": "count(0)",
             "Inputs": [
               {
@@ -1291,7 +1291,7 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
         "Inputs": [
           {
             "OperatorType": "Aggregate",
-            "Variant": "Ordered",
+            "Variant": "Scalar",
             "Aggregates": "count(0) AS s",
             "Inputs": [
               {
@@ -1309,7 +1309,7 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
           },
           {
             "OperatorType": "Aggregate",
-            "Variant": "Ordered",
+            "Variant": "Scalar",
             "Aggregates": "count(0) AS s",
             "Inputs": [
               {
diff --git a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
index 575cc25bd8..37bf043858 100644
--- a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
@@ -443,7 +443,7 @@ Gen4 plan same as above
   "Original": "select id, count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Ordered",
+    "Variant": "Scalar",
     "Aggregates": "count(1)",
     "Inputs": [
       {