Repository: vitessio/vitess
Pull request #9816 (2022-03-02 14:20:29)

Online DDL: cleanup old vtctld+topo flow

## Description

Per https://github.com/vitessio/vitess/issues/6926#issuecomment-817619302, we are good to clean up some code (could have done so in `v13` actually).

This PR removes legacy Online DDL code where `vtctld` took active part in monitoring for schema change requests and communicating those to the tablets/shards. Migrations no longer go through `vtctld`, nor through `topo`. Anything that references that, is removed.

## Related Issue(s)

https://github.com/vitessio/vitess/issues/6926

## Checklist
- [ ] Should this PR be backported?
- [x] Tests were added or are not required
- [ ] 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/cmd/vtctlclient/main.go b/go/cmd/vtctlclient/main.go
index bd8871851f..9722089354 100644
--- a/go/cmd/vtctlclient/main.go
+++ b/go/cmd/vtctlclient/main.go
@@ -42,6 +42,29 @@ var (
 	server        = flag.String("server", "", "server to use for connection")
 )
 
+// checkDeprecations runs quick and dirty checks to see whether any command or flag are deprecated.
+// For any depracated command or flag, the function issues a warning message.
+// this function will change on each Vitess version. Each depracation message should only last a version.
+// VEP-4 will replace the need for this function. See https://github.com/vitessio/enhancements/blob/main/veps/vep-4.md
+func checkDeprecations(args []string) {
+	// utility:
+	findSubstring := func(s string) (arg string, ok bool) {
+		for _, arg := range args {
+			if strings.Contains(arg, s) {
+				return arg, true
+			}
+		}
+		return "", false
+	}
+	if _, ok := findSubstring("ApplySchema"); ok {
+		if arg, ok := findSubstring("ddl_strategy"); ok {
+			if strings.Contains(arg, "-skip-topo") {
+				log.Warning("-skip-topo is deprecated and will be removed in future versions")
+			}
+		}
+	}
+}
+
 func main() {
 	defer exit.Recover()
 
@@ -61,6 +84,8 @@ func main() {
 	ctx, cancel := context.WithTimeout(context.Background(), *actionTimeout)
 	defer cancel()
 
+	checkDeprecations(flag.Args())
+
 	err := vtctlclient.RunCommandAndWait(
 		ctx, *server, flag.Args(),
 		func(e *logutilpb.Event) {
diff --git a/go/vt/schema/ddl_strategy.go b/go/vt/schema/ddl_strategy.go
index 872621ca51..392a9ba056 100644
--- a/go/vt/schema/ddl_strategy.go
+++ b/go/vt/schema/ddl_strategy.go
@@ -29,7 +29,7 @@ var (
 
 const (
 	declarativeFlag        = "declarative"
-	skipTopoFlag           = "skip-topo"
+	skipTopoFlag           = "skip-topo" // legacy. Kept for backwards compatibility, but unused
 	singletonFlag          = "singleton"
 	singletonContextFlag   = "singleton-context"
 	allowZeroInDateFlag    = "allow-zero-in-date"
@@ -155,6 +155,12 @@ func (setting *DDLStrategySetting) IsVreplicationTestSuite() bool {
 	return setting.hasFlag(vreplicationTestSuite)
 }
 
+// IsSkipTopoFlag returns 'true' if strategy options include `-skip-topo`. This flag is deprecated,
+// and this function is temporary in v14 so that we can print a deprecation message.
+func (setting *DDLStrategySetting) IsSkipTopoFlag() bool {
+	return setting.hasFlag(skipTopoFlag)
+}
+
 // RuntimeOptions returns the options used as runtime flags for given strategy, removing any internal hint options
 func (setting *DDLStrategySetting) RuntimeOptions() []string {
 	opts, _ := shlex.Split(setting.Options)
@@ -176,14 +182,6 @@ func (setting *DDLStrategySetting) RuntimeOptions() []string {
 	return validOpts
 }
 
-// IsSkipTopo suggests that DDL should apply to tables bypassing global topo request
-func (setting *DDLStrategySetting) IsSkipTopo() bool {
-	// Vitess 11 introduced the flag -skip-topo. starting Vitess 12 the flag is _always_ considered 'true'.
-	// Ideally the flag should be gone, but for backwards compatibility we allow users to still specify it
-	// (and we stil ignore the value, it's always set to true)
-	return true
-}
-
 // ToString returns a simple string representation of this instance
 func (setting *DDLStrategySetting) ToString() string {
 	return fmt.Sprintf("DDLStrategySetting: strategy=%v, options=%s", setting.Strategy, setting.Options)
diff --git a/go/vt/schema/online_ddl.go b/go/vt/schema/online_ddl.go
index 0dd6059f2b..2e48a60045 100644
--- a/go/vt/schema/online_ddl.go
+++ b/go/vt/schema/online_ddl.go
@@ -17,7 +17,6 @@ limitations under the License.
 package schema
 
 import (
-	"context"
 	"encoding/hex"
 	"encoding/json"
 	"errors"
@@ -29,7 +28,6 @@ import (
 
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/sqlparser"
-	"vitess.io/vitess/go/vt/topo"
 	"vitess.io/vitess/go/vt/vterrors"
 )
 
@@ -133,19 +131,6 @@ func FromJSON(bytes []byte) (*OnlineDDL, error) {
 	return onlineDDL, err
 }
 
-// ReadTopo reads a OnlineDDL object from given topo connection
-func ReadTopo(ctx context.Context, conn topo.Conn, entryPath string) (*OnlineDDL, error) {
-	bytes, _, err := conn.Get(ctx, entryPath)
-	if err != nil {
-		return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "ReadTopo Get %s error: %s", entryPath, err.Error())
-	}
-	onlineDDL, err := FromJSON(bytes)
-	if err != nil {
-		return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "ReadTopo unmarshal %s error: %s", entryPath, err.Error())
-	}
-	return onlineDDL, nil
-}
-
 // ParseOnlineDDLStatement parses the given SQL into a statement and returns the action type of the DDL statement, or error
 // if the statement is not a DDL
 func ParseOnlineDDLStatement(sql string) (ddlStmt sqlparser.DDLStatement, action sqlparser.DDLAction, err error) {
@@ -240,19 +225,16 @@ func NewOnlineDDL(keyspace string, table string, sql string, ddlStrategySetting
 		encodeDirective := func(directive string) string {
 			return strconv.Quote(hex.EncodeToString([]byte(directive)))
 		}
-		var comments sqlparser.Comments
-		if ddlStrategySetting.IsSkipTopo() {
-			comments = sqlparser.Comments{
-				fmt.Sprintf(`/*vt+ uuid=%s context=%s table=%s strategy=%s options=%s */`,
-					encodeDirective(onlineDDLUUID),
-					encodeDirective(migrationContext),
-					encodeDirective(table),
-					encodeDirective(string(ddlStrategySetting.Strategy)),
-					encodeDirective(ddlStrategySetting.Options),
-				)}
-			if uuid, err := legacyParseRevertUUID(sql); err == nil {
-				sql = fmt.Sprintf("revert vitess_migration '%s'", uuid)
-			}
+		comments := sqlparser.Comments{
+			fmt.Sprintf(`/*vt+ uuid=%s context=%s table=%s strategy=%s options=%s */`,
+				encodeDirective(onlineDDLUUID),
+				encodeDirective(migrationContext),
+				encodeDirective(table),
+				encodeDirective(string(ddlStrategySetting.Strategy)),
+				encodeDirective(ddlStrategySetting.Options),
+			)}
+		if uuid, err := legacyParseRevertUUID(sql); err == nil {
+			sql = fmt.Sprintf("revert vitess_migration '%s'", uuid)
 		}
 
 		stmt, err := sqlparser.Parse(sql)
@@ -473,22 +455,6 @@ func (onlineDDL *OnlineDDL) ToString() string {
 	return fmt.Sprintf("OnlineDDL: keyspace=%s, table=%s, sql=%s", onlineDDL.Keyspace, onlineDDL.Table, onlineDDL.SQL)
 }
 
-// WriteTopo writes this online DDL to given topo connection, based on basePath and and this DDL's UUID
-func (onlineDDL *OnlineDDL) WriteTopo(ctx context.Context, conn topo.Conn, basePath string) error {
-	if onlineDDL.UUID == "" {
-		return vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "onlineDDL UUID not found; keyspace=%s, sql=%s", onlineDDL.Keyspace, onlineDDL.SQL)
-	}
-	bytes, err := onlineDDL.ToJSON()
-	if err != nil {
-		return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "onlineDDL marshall error:%s, keyspace=%s, sql=%s", err.Error(), onlineDDL.Keyspace, onlineDDL.SQL)
-	}
-	_, err = conn.Create(ctx, fmt.Sprintf("%s/%s", basePath, onlineDDL.UUID), bytes)
-	if err != nil {
-		return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "onlineDDL topo create error:%s, keyspace=%s, sql=%s", err.Error(), onlineDDL.Keyspace, onlineDDL.SQL)
-	}
-	return nil
-}
-
 // GetGCUUID gets this OnlineDDL UUID in GC UUID format
 func (onlineDDL *OnlineDDL) GetGCUUID() string {
 	return OnlineDDLToGCUUID(onlineDDL.UUID)
diff --git a/go/vt/schemamanager/tablet_executor.go b/go/vt/schemamanager/tablet_executor.go
index b70a807222..3edc6b5cd6 100644
--- a/go/vt/schemamanager/tablet_executor.go
+++ b/go/vt/schemamanager/tablet_executor.go
@@ -361,28 +361,6 @@ func (exec *TabletExecutor) Execute(ctx context.Context, sqls []string) *Execute
 	return &execResult
 }
 
-// executeOnlineDDL submits an online DDL request; this runs on topo, not on tablets, and is a quick operation.
-func (exec *TabletExecutor) executeOnlineDDL(
-	ctx context.Context, execResult *ExecuteResult, onlineDDL *schema.OnlineDDL,
-) {
-	if exec.ddlStrategySetting == nil || exec.ddlStrategySetting.Strategy.IsDirect() {
-		execResult.ExecutorErr = "Not an online DDL strategy"
-		return
-	}
-	conn, err := exec.ts.ConnForCell(ctx, topo.GlobalCell)
-	if err != nil {
-		execResult.ExecutorErr = fmt.Sprintf("online DDL ConnForCell error:%s", err.Error())
-		return
-	}
-	err = onlineDDL.WriteTopo(ctx, conn, schema.MigrationRequestsPath())
-	if err != nil {
-		execResult.ExecutorErr = err.Error()
-	}
-	execResult.UUIDs = append(execResult.UUIDs, onlineDDL.UUID)
-	exec.logger.Infof("UUID=%+v", onlineDDL.UUID)
-	exec.logger.Printf("%s\n", onlineDDL.UUID)
-}
-
 // executeOnAllTablets runs a query on all tablets, synchronously. This can be a long running operation.
 func (exec *TabletExecutor) executeOnAllTablets(ctx context.Context, execResult *ExecuteResult, sql string, viaQueryService bool) {
 	var wg sync.WaitGroup
diff --git a/go/vt/vtctl/vtctl.go b/go/vt/vtctl/vtctl.go
index d92ec3deee..cc29214e16 100644
--- a/go/vt/vtctl/vtctl.go
+++ b/go/vt/vtctl/vtctl.go
@@ -3263,6 +3263,12 @@ func commandApplySchema(ctx context.Context, wr *wrangler.Wrangler, subFlags *fl
 		return fmt.Errorf("the <keyspace> argument is required for the commandApplySchema command")
 	}
 
+	// v14 deprecates `-skip-topo` flag. This check will be removed in v15
+	if settings, _ := schema.ParseDDLStrategy(*ddlStrategy); settings != nil && settings.IsSkipTopoFlag() {
+		deprecationMessage := `-skip-topo flag is deprecated and will be removed in v15`
+		log.Warningf(deprecationMessage)
+	}
+
 	keyspace := subFlags.Arg(0)
 	change, err := getFileParam(*sql, *sqlFile, "sql")
 	if err != nil {
diff --git a/go/vt/vtctld/schema.go b/go/vt/vtctld/schema.go
deleted file mode 100644
index 6b9c559f80..0000000000
--- a/go/vt/vtctld/schema.go
+++ /dev/null
@@ -1,196 +0,0 @@
-/*
-Copyright 2019 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 vtctld
-
-import (
-	"flag"
-	"fmt"
-	"sync"
-	"time"
-
-	"context"
-
-	"vitess.io/vitess/go/sqltypes"
-	"vitess.io/vitess/go/timer"
-	"vitess.io/vitess/go/vt/log"
-	"vitess.io/vitess/go/vt/logutil"
-	querypb "vitess.io/vitess/go/vt/proto/query"
-	"vitess.io/vitess/go/vt/schema"
-	"vitess.io/vitess/go/vt/servenv"
-	"vitess.io/vitess/go/vt/sqlparser"
-	"vitess.io/vitess/go/vt/topo"
-	"vitess.io/vitess/go/vt/vttablet/tmclient"
-	"vitess.io/vitess/go/vt/wrangler"
-)
-
-var (
-	migrationCheckTicks *timer.Timer
-	onlineDDLOnce       sync.Once
-)
-
-var (
-	migrationCheckInterval = flag.Duration("online_ddl_check_interval", time.Minute, "interval polling for new online DDL requests")
-)
-
-func initSchemaManager(ts *topo.Server) {
-	tmClient := tmclient.NewTabletManagerClient()
-	migrationCheckTicks = timer.NewTimer(*migrationCheckInterval)
-
-	runMigrationRequestChecks(ts, tmClient)
-}
-
-func runMigrationRequestChecks(ts *topo.Server, tmClient tmclient.TabletManagerClient) {
-	ctx, cancel := context.WithCancel(context.Background())
-	migrationCheckTicks.Start(func() { onMigrationCheckTick(ctx, ts, tmClient) })
-
-	go func() {
-		<-ctx.Done()
-		migrationCheckTicks.Stop()
-	}()
-
-	// Running cancel on OnTermSync will cancel the context of any
-	// running workflow inside vtctld. They may still checkpoint
-	// if they want to.
-	servenv.OnTermSync(cancel)
-}
-
-func reviewMigrationRequest(ctx context.Context, ts *topo.Server, tmClient tmclient.TabletManagerClient, conn topo.Conn, uuid string) error {
-	if !schema.IsOnlineDDLUUID(uuid) {
-		// Just some other entry in this path, e.g. a sentry or a placeholder.
-		return nil
-	}
-	entryPath := fmt.Sprintf("%s/%s", schema.MigrationRequestsPath(), uuid)
-	onlineDDL, err := schema.ReadTopo(ctx, conn, entryPath)
-	if err != nil {
-		return err
-	}
-	_, actionStr, err := onlineDDL.GetActionStr()
-	if err != nil {
-		return err
-	}
-	log.Infof("Found schema migration request: %+v", onlineDDL)
-
-	onlineDDL.Status = schema.OnlineDDLStatusQueued
-
-	logstream := logutil.NewMemoryLogger()
-	wr := wrangler.New(logstream, ts, tmClient)
-
-	sqlInsertSchemaMigration := `INSERT IGNORE INTO %s.schema_migrations (
-		migration_uuid,
-		keyspace,
-		shard,
-		mysql_schema,
-		mysql_table,
-		migration_statement,
-		strategy,
-		options,
-		ddl_action,
-		requested_timestamp,
-		migration_context,
-		migration_status
-	) VALUES (
-		%a, %a, %a, %a, %a, %a, %a, %a, %a, FROM_UNIXTIME(%a), %a, %a
-	)`
-	parsed := sqlparser.BuildParsedQuery(sqlInsertSchemaMigration, "_vt",
-		":migration_uuid",
-		":keyspace",
-		":shard",
-		":mysql_schema",
-		":mysql_table",
-		":migration_statement",
-		":strategy",
-		":options",
-		":ddl_action",
-		":requested_timestamp",
-		":migration_context",
-		":migration_status",
-	)
-	bindVars := map[string]*querypb.BindVariable{
-		"migration_uuid":      sqltypes.StringBindVariable(onlineDDL.UUID),
-		"keyspace":            sqltypes.StringBindVariable(onlineDDL.Keyspace),
-		"shard":               sqltypes.StringBindVariable(""),
-		"mysql_schema":        sqltypes.StringBindVariable(""),
-		"mysql_table":         sqltypes.StringBindVariable(onlineDDL.Table),
-		"migration_statement": sqltypes.StringBindVariable(onlineDDL.SQL),
-		"strategy":            sqltypes.StringBindVariable(string(onlineDDL.Strategy)),
-		"options":             sqltypes.StringBindVariable(onlineDDL.Options),
-		"ddl_action":          sqltypes.StringBindVariable(actionStr),
-		"requested_timestamp": sqltypes.Int64BindVariable(onlineDDL.RequestTimeSeconds()),
-		"migration_context":   sqltypes.StringBindVariable(onlineDDL.MigrationContext),
-		"migration_status":    sqltypes.StringBindVariable(string(onlineDDL.Status)),
-	}
-
-	bound, err := parsed.GenerateQuery(bindVars, nil)
-	if err != nil {
-		return err
-	}
-
-	_, err = wr.VExecResult(ctx, onlineDDL.UUID, onlineDDL.Keyspace, bound, false)
-	if err != nil {
-		return err
-	}
-
-	if err := onlineDDL.WriteTopo(ctx, conn, schema.MigrationQueuedPath()); err != nil {
-		return fmt.Errorf("unable to write reviewed migration: %+v, error: %s", onlineDDL, err)
-	}
-	if err := conn.Delete(ctx, entryPath, nil); err != nil {
-		return fmt.Errorf("unable to delete %+v, error: %s", entryPath, err)
-	}
-	return nil
-}
-
-func reviewMigrationRequests(ctx context.Context, ts *topo.Server, tmClient tmclient.TabletManagerClient, conn topo.Conn) error {
-	entries, err := conn.ListDir(ctx, schema.MigrationRequestsPath(), true)
-	if err != nil {
-		log.Errorf("vtctld.reviewMigrationRequests listDir error: %s", err.Error())
-		return err
-	}
-
-	for _, entry := range entries {
-		if err := reviewMigrationRequest(ctx, ts, tmClient, conn, entry.Name); err != nil {
-			log.Errorf("vtctld.reviewMigrationRequest %s error: %s", entry.Name, err.Error())
-			continue
-		}
-	}
-	return nil
-}
-
-func onMigrationCheckTick(ctx context.Context, ts *topo.Server, tmClient tmclient.TabletManagerClient) {
-	conn, err := ts.ConnForCell(ctx, topo.GlobalCell)
-	if err != nil {
-		log.Errorf("vtctld.onMigrationCheckTick ConnForCell error: %s", err.Error())
-		return
-	}
-
-	onlineDDLOnce.Do(func() {
-		// This creates the directory schema.MigrationRequestsPath(), once.
-		// From now on, we can ListDir() n that directory without errors, even if no migration has ever been created.
-		_, err := conn.Create(ctx, fmt.Sprintf("%s/sentry", schema.MigrationRequestsPath()), []byte{})
-		if err != nil && !topo.IsErrType(err, topo.NodeExists) {
-			log.Errorf("vtctld.onMigrationCheckTick Create sentry error: %s", err.Error())
-		}
-	})
-
-	lockDescriptor, err := conn.Lock(ctx, schema.MigrationRequestsPath(), "vtctld.onMigrationCheckTick")
-	if err != nil {
-		log.Errorf("vtctld.onMigrationCheckTick ConnForCell error: %s", err.Error())
-		return
-	}
-	defer lockDescriptor.Unlock(ctx)
-
-	reviewMigrationRequests(ctx, ts, tmClient, conn)
-}
diff --git a/go/vt/vtctld/vtctld.go b/go/vt/vtctld/vtctld.go
index f72ae1a62c..5d5c3483d4 100644
--- a/go/vt/vtctld/vtctld.go
+++ b/go/vt/vtctld/vtctld.go
@@ -162,9 +162,6 @@ func InitVtctld(ts *topo.Server) error {
 	// Init workflow manager.
 	initWorkflowManager(ts)
 
-	// Init online DDL schema manager
-	initSchemaManager(ts)
-
 	// Setup reverse proxy for all vttablets through /vttablet/.
 	initVTTabletRedirection(ts)
 
diff --git a/go/vt/vtgate/engine/online_ddl.go b/go/vt/vtgate/engine/online_ddl.go
index bfc3c6dc5d..574873978f 100644
--- a/go/vt/vtgate/engine/online_ddl.go
+++ b/go/vt/vtgate/engine/online_ddl.go
@@ -89,23 +89,16 @@ func (v *OnlineDDL) TryExecute(vcursor VCursor, bindVars map[string]*query.BindV
 		return result, err
 	}
 	for _, onlineDDL := range onlineDDLs {
-		if onlineDDL.StrategySetting().IsSkipTopo() {
-			// Go directly to tablets, much like Send primitive does
-			s := Send{
-				Keyspace:          v.Keyspace,
-				TargetDestination: v.TargetDestination,
-				Query:             onlineDDL.SQL,
-				IsDML:             false,
-				SingleShardOnly:   false,
-			}
-			if _, err := vcursor.ExecutePrimitive(&s, bindVars, wantfields); err != nil {
-				return result, err
-			}
-		} else {
-			// Submit a request entry in topo. vtctld will take it from there
-			if err := vcursor.SubmitOnlineDDL(onlineDDL); err != nil {
-				return result, err
-			}
+		// Go directly to tablets, much like Send primitive does
+		s := Send{
+			Keyspace:          v.Keyspace,
+			TargetDestination: v.TargetDestination,
+			Query:             onlineDDL.SQL,
+			IsDML:             false,
+			SingleShardOnly:   false,
+		}
+		if _, err := vcursor.ExecutePrimitive(&s, bindVars, wantfields); err != nil {
+			return result, err
 		}
 		result.Rows = append(result.Rows, []sqltypes.Value{
 			sqltypes.NewVarChar(onlineDDL.UUID),
diff --git a/go/vt/vtgate/engine/ordered_aggregate.go b/go/vt/vtgate/engine/ordered_aggregate.go
index 1c6b82d592..90aa1e6222 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: convertFields(result.Fields, oa.PreProcess, oa.Aggregates),
+		Fields: oa.convertFields(result.Fields),
 		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 = convertRow(row, oa.PreProcess, oa.Aggregates)
+			current, curDistincts = oa.convertRow(row)
 			continue
 		}
 		equal, err := oa.keysEqual(current, row, oa.Collations)
@@ -241,18 +241,28 @@ func (oa *OrderedAggregate) execute(vcursor VCursor, bindVars map[string]*queryp
 		}
 
 		if equal {
-			current, curDistincts, err = merge(result.Fields, current, row, curDistincts, oa.Collations, oa.Aggregates)
+			current, curDistincts, err = oa.merge(result.Fields, current, row, curDistincts, oa.Collations)
 			if err != nil {
 				return nil, err
 			}
 			continue
 		}
 		out.Rows = append(out.Rows, current)
-		current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates)
+		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)
 	}
 
 	if current != nil {
-		final, err := convertFinal(current, oa.Aggregates)
+		final, err := oa.convertFinal(current)
 		if err != nil {
 			return nil, err
 		}
@@ -273,7 +283,7 @@ func (oa *OrderedAggregate) TryStreamExecute(vcursor VCursor, bindVars map[strin
 
 	err := vcursor.StreamExecutePrimitive(oa.Input, bindVars, wantfields, func(qr *sqltypes.Result) error {
 		if len(qr.Fields) != 0 {
-			fields = convertFields(qr.Fields, oa.PreProcess, oa.Aggregates)
+			fields = oa.convertFields(qr.Fields)
 			if err := cb(&sqltypes.Result{Fields: fields}); err != nil {
 				return err
 			}
@@ -281,7 +291,7 @@ func (oa *OrderedAggregate) TryStreamExecute(vcursor VCursor, bindVars map[strin
 		// This code is similar to the one in Execute.
 		for _, row := range qr.Rows {
 			if current == nil {
-				current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates)
+				current, curDistincts = oa.convertRow(row)
 				continue
 			}
 
@@ -291,7 +301,7 @@ func (oa *OrderedAggregate) TryStreamExecute(vcursor VCursor, bindVars map[strin
 			}
 
 			if equal {
-				current, curDistincts, err = merge(fields, current, row, curDistincts, oa.Collations, oa.Aggregates)
+				current, curDistincts, err = oa.merge(fields, current, row, curDistincts, oa.Collations)
 				if err != nil {
 					return err
 				}
@@ -300,7 +310,7 @@ func (oa *OrderedAggregate) TryStreamExecute(vcursor VCursor, bindVars map[strin
 			if err := cb(&sqltypes.Result{Rows: [][]sqltypes.Value{current}}); err != nil {
 				return err
 			}
-			current, curDistincts = convertRow(row, oa.PreProcess, oa.Aggregates)
+			current, curDistincts = oa.convertRow(row)
 		}
 		return nil
 	})
@@ -316,11 +326,11 @@ func (oa *OrderedAggregate) TryStreamExecute(vcursor VCursor, bindVars map[strin
 	return nil
 }
 
-func convertFields(fields []*querypb.Field, preProcess bool, aggrs []*AggregateParams) []*querypb.Field {
-	if !preProcess {
+func (oa *OrderedAggregate) convertFields(fields []*querypb.Field) []*querypb.Field {
+	if !oa.PreProcess {
 		return fields
 	}
-	for _, aggr := range aggrs {
+	for _, aggr := range oa.Aggregates {
 		if !aggr.preProcess() {
 			continue
 		}
@@ -335,13 +345,13 @@ func convertFields(fields []*querypb.Field, preProcess bool, aggrs []*AggregateP
 	return fields
 }
 
-func convertRow(row []sqltypes.Value, preProcess bool, aggregates []*AggregateParams) (newRow []sqltypes.Value, curDistincts []sqltypes.Value) {
-	if !preProcess {
+func (oa *OrderedAggregate) convertRow(row []sqltypes.Value) (newRow []sqltypes.Value, curDistincts []sqltypes.Value) {
+	if !oa.PreProcess {
 		return row, nil
 	}
 	newRow = append(newRow, row...)
-	curDistincts = make([]sqltypes.Value, len(aggregates))
-	for index, aggr := range aggregates {
+	curDistincts = make([]sqltypes.Value, len(oa.Aggregates))
+	for index, aggr := range oa.Aggregates {
 		switch aggr.Opcode {
 		case AggregateCountDistinct:
 			curDistincts[index] = findComparableCurrentDistinct(row, aggr)
@@ -388,7 +398,7 @@ func (oa *OrderedAggregate) GetFields(vcursor VCursor, bindVars map[string]*quer
 	if err != nil {
 		return nil, err
 	}
-	qr = &sqltypes.Result{Fields: convertFields(qr.Fields, oa.PreProcess, oa.Aggregates)}
+	qr = &sqltypes.Result{Fields: oa.convertFields(qr.Fields)}
 	return qr.Truncate(oa.TruncateColumnCount), nil
 }
 
@@ -424,15 +434,9 @@ func (oa *OrderedAggregate) keysEqual(row1, row2 []sqltypes.Value, colls map[int
 	return true, nil
 }
 
-func merge(
-	fields []*querypb.Field,
-	row1, row2 []sqltypes.Value,
-	curDistincts []sqltypes.Value,
-	colls map[int]collations.ID,
-	aggregates []*AggregateParams,
-) ([]sqltypes.Value, []sqltypes.Value, error) {
+func (oa *OrderedAggregate) merge(fields []*querypb.Field, row1, row2 []sqltypes.Value, curDistincts []sqltypes.Value, colls map[int]collations.ID) ([]sqltypes.Value, []sqltypes.Value, error) {
 	result := sqltypes.CopyRow(row1)
-	for index, aggr := range aggregates {
+	for index, aggr := range oa.Aggregates {
 		if aggr.isDistinct() {
 			if row2[aggr.KeyCol].IsNull() {
 				continue
@@ -488,6 +492,19 @@ func merge(
 	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
@@ -530,9 +547,9 @@ func (oa *OrderedAggregate) description() PrimitiveDescription {
 	}
 }
 
-func convertFinal(current []sqltypes.Value, aggregates []*AggregateParams) ([]sqltypes.Value, error) {
+func (oa *OrderedAggregate) convertFinal(current []sqltypes.Value) ([]sqltypes.Value, error) {
 	result := sqltypes.CopyRow(current)
-	for _, aggr := range aggregates {
+	for _, aggr := range oa.Aggregates {
 		switch aggr.Opcode {
 		case AggregateGtid:
 			vgtid := &binlogdatapb.VGtid{}
diff --git a/go/vt/vtgate/engine/primitive.go b/go/vt/vtgate/engine/primitive.go
index 43d8953279..493a784c3f 100644
--- a/go/vt/vtgate/engine/primitive.go
+++ b/go/vt/vtgate/engine/primitive.go
@@ -27,7 +27,6 @@ import (
 	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/key"
-	"vitess.io/vitess/go/vt/schema"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/srvtopo"
 	"vitess.io/vitess/go/vt/vtgate/vindexes"
@@ -92,8 +91,6 @@ type (
 
 		ExecuteVSchema(keyspace string, vschemaDDL *sqlparser.AlterVschema) error
 
-		SubmitOnlineDDL(onlineDDl *schema.OnlineDDL) error
-
 		Session() SessionActions
 
 		ConnCollation() collations.ID
diff --git a/go/vt/vtgate/engine/revert_migration.go b/go/vt/vtgate/engine/revert_migration.go
index 98a077883e..2adffd36b2 100644
--- a/go/vt/vtgate/engine/revert_migration.go
+++ b/go/vt/vtgate/engine/revert_migration.go
@@ -93,22 +93,15 @@ func (v *RevertMigration) TryExecute(vcursor VCursor, bindVars map[string]*query
 		return result, err
 	}
 
-	if ddlStrategySetting.IsSkipTopo() {
-		s := Send{
-			Keyspace:          v.Keyspace,
-			TargetDestination: v.TargetDestination,
-			Query:             onlineDDL.SQL,
-			IsDML:             false,
-			SingleShardOnly:   false,
-		}
-		if _, err := vcursor.ExecutePrimitive(&s, bindVars, wantfields); err != nil {
-			return result, err
-		}
-	} else {
-		// Submit a request entry in topo. vtctld will take it from there
-		if err := vcursor.SubmitOnlineDDL(onlineDDL); err != nil {
-			return result, err
-		}
+	s := Send{
+		Keyspace:          v.Keyspace,
+		TargetDestination: v.TargetDestination,
+		Query:             onlineDDL.SQL,
+		IsDML:             false,
+		SingleShardOnly:   false,
+	}
+	if _, err := vcursor.ExecutePrimitive(&s, bindVars, wantfields); err != nil {
+		return result, err
 	}
 	result.Rows = append(result.Rows, []sqltypes.Value{
 		sqltypes.NewVarChar(onlineDDL.UUID),
diff --git a/go/vt/vtgate/engine/scalar_aggregation.go b/go/vt/vtgate/engine/scalar_aggregation.go
deleted file mode 100644
index 10884ae5f8..0000000000
--- a/go/vt/vtgate/engine/scalar_aggregation.go
+++ /dev/null
@@ -1,193 +0,0 @@
-/*
-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/planbuilder/grouping.go b/go/vt/vtgate/planbuilder/grouping.go
index 2bf58dcfa4..84335ebc61 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.groupByKeys = append(node.groupByKeys, &engine.GroupByParams{KeyCol: colNumber, WeightStringCol: -1, FromGroupBy: true})
+			node.eaggr.GroupByKeys = append(node.eaggr.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.groupByKeys = append(node.groupByKeys, &engine.GroupByParams{KeyCol: i, WeightStringCol: -1, FromGroupBy: false})
+			node.eaggr.GroupByKeys = append(node.eaggr.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 aedd72c683..d12cd7bcfa 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.truncateColumnCount = hp.sel.GetColumnCount()
+		p.eaggr.SetTruncateColumnCount(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.aggregates {
+		for _, aggregate := range node.eaggr.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.groupByKeys {
+		for _, key := range node.eaggr.GroupByKeys {
 			if sqlparser.EqualsExpr(key.Expr, expr.Expr) {
 				return key.KeyCol, false, nil
 			}
@@ -473,11 +473,14 @@ 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
@@ -523,7 +526,7 @@ func (hp *horizonPlanning) planAggregations(ctx *plancontext.PlanningContext, pl
 		}
 		param.Col = offset
 		param.Expr = fExpr
-		oa.aggregates = append(oa.aggregates, param)
+		oa.eaggr.Aggregates = append(oa.eaggr.Aggregates, param)
 	}
 
 	for _, groupExpr := range hp.qp.GroupByExprs {
@@ -577,7 +580,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,
@@ -606,11 +609,11 @@ func (hp *horizonPlanning) createPushExprAndAlias(
 			opcode = engine.AggregateSumDistinct
 		}
 
-		oa.preProcess = true
+		oa.eaggr.PreProcess = true
 		by := abstract.GroupBy{
 			Inner:             innerAliased.Expr,
 			WeightStrExpr:     innerAliased.Expr,
-			DistinctAggrIndex: len(oa.aggregates) + 1,
+			DistinctAggrIndex: len(oa.eaggr.Aggregates) + 1,
 		}
 		hp.qp.GroupByExprs = append(hp.qp.GroupByExprs, by)
 	}
@@ -657,11 +660,11 @@ func planGroupByGen4(ctx *plancontext.PlanningContext, groupExpr abstract.GroupB
 			return err
 		}
 		if groupExpr.DistinctAggrIndex == 0 {
-			node.groupByKeys = append(node.groupByKeys, &engine.GroupByParams{KeyCol: keyCol, WeightStringCol: wsOffset, Expr: groupExpr.WeightStrExpr, CollationID: ctx.SemTable.CollationForExpr(groupExpr.Inner)})
+			node.eaggr.GroupByKeys = append(node.eaggr.GroupByKeys, &engine.GroupByParams{KeyCol: keyCol, WeightStringCol: wsOffset, Expr: groupExpr.WeightStrExpr, CollationID: ctx.SemTable.CollationForExpr(groupExpr.Inner)})
 		} else {
 			if wsOffset != -1 {
-				node.aggregates[groupExpr.DistinctAggrIndex-1].WAssigned = true
-				node.aggregates[groupExpr.DistinctAggrIndex-1].WCol = wsOffset
+				node.eaggr.Aggregates[groupExpr.DistinctAggrIndex-1].WAssigned = true
+				node.eaggr.Aggregates[groupExpr.DistinctAggrIndex-1].WCol = wsOffset
 			}
 		}
 		err = planGroupByGen4(ctx, groupExpr, node.input, wsOffset != -1)
@@ -946,7 +949,7 @@ func createMemorySortPlanOnAggregation(plan *orderedAggregate, orderExprs []abst
 
 		collationID := collations.Unknown
 		if woffset != -1 {
-			collationID = plan.groupByKeys[idx].CollationID
+			collationID = plan.eaggr.GroupByKeys[idx].CollationID
 		}
 		ms.eMemorySort.OrderBy = append(ms.eMemorySort.OrderBy, engine.OrderByParams{
 			Col:               offset,
@@ -960,12 +963,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.groupByKeys {
+	for idx, key := range plan.eaggr.GroupByKeys {
 		if sqlparser.EqualsExpr(order.WeightStrExpr, key.Expr) {
 			return key.KeyCol, key.WeightStringCol, idx, true
 		}
 	}
-	for idx, aggregate := range plan.aggregates {
+	for idx, aggregate := range plan.eaggr.Aggregates {
 		if sqlparser.EqualsExpr(order.WeightStrExpr, aggregate.Expr) {
 			return aggregate.Col, -1, idx, true
 		}
@@ -1038,11 +1041,14 @@ 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()
@@ -1050,20 +1056,20 @@ func (hp *horizonPlanning) planDistinctOA(semTable *semantics.SemTable, currPlan
 			return nil, err
 		}
 		found := false
-		for _, grpParam := range currPlan.groupByKeys {
+		for _, grpParam := range currPlan.eaggr.GroupByKeys {
 			if sqlparser.EqualsExpr(expr, grpParam.Expr) {
 				found = true
-				oa.groupByKeys = append(oa.groupByKeys, grpParam)
+				eaggr.GroupByKeys = append(eaggr.GroupByKeys, grpParam)
 				break
 			}
 		}
 		if found {
 			continue
 		}
-		for _, aggrParam := range currPlan.aggregates {
+		for _, aggrParam := range currPlan.eaggr.Aggregates {
 			if sqlparser.EqualsExpr(expr, aggrParam.Expr) {
 				found = true
-				oa.groupByKeys = append(oa.groupByKeys, &engine.GroupByParams{KeyCol: aggrParam.Col, WeightStringCol: -1, CollationID: semTable.CollationForExpr(expr)})
+				eaggr.GroupByKeys = append(eaggr.GroupByKeys, &engine.GroupByParams{KeyCol: aggrParam.Col, WeightStringCol: -1, CollationID: semTable.CollationForExpr(expr)})
 				break
 			}
 		}
@@ -1075,8 +1081,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 {
@@ -1101,7 +1107,7 @@ func (hp *horizonPlanning) addDistinct(ctx *plancontext.PlanningContext, plan lo
 			return nil, err
 		}
 		grpParam.WeightStringCol = wOffset
-		groupByKeys = append(groupByKeys, grpParam)
+		eaggr.GroupByKeys = append(eaggr.GroupByKeys, grpParam)
 
 		orderExprs = append(orderExprs, abstract.OrderBy{
 			Inner:         &sqlparser.Order{Expr: inner},
@@ -1116,8 +1122,9 @@ func (hp *horizonPlanning) addDistinct(ctx *plancontext.PlanningContext, plan lo
 		resultsBuilder: resultsBuilder{
 			logicalPlanCommon: newBuilderCommon(innerPlan),
 			weightStrings:     make(map[*resultColumn]int),
+			truncater:         eaggr,
 		},
-		groupByKeys: groupByKeys,
+		eaggr: eaggr,
 	}
 	return oa, nil
 }
diff --git a/go/vt/vtgate/planbuilder/ordered_aggregate.go b/go/vt/vtgate/planbuilder/ordered_aggregate.go
index de1fff1c97..8cc037d6ac 100644
--- a/go/vt/vtgate/planbuilder/ordered_aggregate.go
+++ b/go/vt/vtgate/planbuilder/ordered_aggregate.go
@@ -63,19 +63,7 @@ var _ logicalPlan = (*orderedAggregate)(nil)
 type orderedAggregate struct {
 	resultsBuilder
 	extraDistinct *sqlparser.ColName
-
-	// 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
+	eaggr         *engine.OrderedAggregate
 }
 
 // checkAggregates analyzes the select expression for aggregates. If it determines
@@ -141,9 +129,11 @@ func (pb *primitiveBuilder) checkAggregates(sel *sqlparser.Select) error {
 	}
 
 	// We need an aggregator primitive.
-	oa := &orderedAggregate{}
-	oa.resultsBuilder = newResultsBuilder(rb, oa)
-	pb.plan = oa
+	eaggr := &engine.OrderedAggregate{}
+	pb.plan = &orderedAggregate{
+		resultsBuilder: newResultsBuilder(rb, eaggr),
+		eaggr:          eaggr,
+	}
 	pb.plan.Reorder(0)
 	return nil
 }
@@ -225,36 +215,8 @@ func findAlias(colname *sqlparser.ColName, selects sqlparser.SelectExprs) sqlpar
 
 // Primitive implements the logicalPlan interface
 func (oa *orderedAggregate) Primitive() engine.Primitive {
-	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,
-	}
+	oa.eaggr.Input = oa.input.Primitive()
+	return oa.eaggr
 }
 
 func (oa *orderedAggregate) pushAggr(pb *primitiveBuilder, expr *sqlparser.AliasedExpr, origin logicalPlan) (rc *resultColumn, colNumber int, err error) {
@@ -283,7 +245,7 @@ func (oa *orderedAggregate) pushAggr(pb *primitiveBuilder, expr *sqlparser.Alias
 			return nil, 0, err
 		}
 		oa.extraDistinct = col
-		oa.preProcess = true
+		oa.eaggr.PreProcess = true
 		var alias string
 		if expr.As.IsEmpty() {
 			alias = sqlparser.String(expr.Expr)
@@ -296,7 +258,7 @@ func (oa *orderedAggregate) pushAggr(pb *primitiveBuilder, expr *sqlparser.Alias
 		case engine.AggregateSum:
 			opcode = engine.AggregateSumDistinct
 		}
-		oa.aggregates = append(oa.aggregates, &engine.AggregateParams{
+		oa.eaggr.Aggregates = append(oa.eaggr.Aggregates, &engine.AggregateParams{
 			Opcode: opcode,
 			Col:    innerCol,
 			Alias:  alias,
@@ -307,7 +269,7 @@ func (oa *orderedAggregate) pushAggr(pb *primitiveBuilder, expr *sqlparser.Alias
 			return nil, 0, err
 		}
 		pb.plan = newBuilder
-		oa.aggregates = append(oa.aggregates, &engine.AggregateParams{
+		oa.eaggr.Aggregates = append(oa.eaggr.Aggregates, &engine.AggregateParams{
 			Opcode: opcode,
 			Col:    innerCol,
 		})
@@ -352,7 +314,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.groupByKeys {
+	for i, gbk := range oa.eaggr.GroupByKeys {
 		rc := oa.resultColumns[gbk.KeyCol]
 		if sqltypes.IsText(rc.column.typ) {
 			weightcolNumber, err := oa.input.SupplyWeightString(gbk.KeyCol, gbk.FromGroupBy)
@@ -364,31 +326,38 @@ func (oa *orderedAggregate) Wireup(plan logicalPlan, jt *jointab) error {
 				return err
 			}
 			oa.weightStrings[rc] = weightcolNumber
-			oa.groupByKeys[i].WeightStringCol = weightcolNumber
-			oa.groupByKeys[i].KeyCol = weightcolNumber
-			oa.truncateColumnCount = len(oa.resultColumns)
+			oa.eaggr.GroupByKeys[i].WeightStringCol = weightcolNumber
+			oa.eaggr.GroupByKeys[i].KeyCol = weightcolNumber
+			oa.eaggr.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.aggregates {
+	for _, aggr := range oa.eaggr.Aggregates {
 		outputCols[aggr.Col] = &sqlparser.AliasedExpr{Expr: aggr.Expr, As: sqlparser.NewColIdent(aggr.Alias)}
 	}
-	if oa.truncateColumnCount > 0 {
-		return outputCols[:oa.truncateColumnCount]
+	if oa.eaggr.TruncateColumnCount > 0 {
+		return outputCols[:oa.eaggr.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 778316bb16..db711f0e72 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.groupByKeys))
+	referenced := make([]bool, len(oa.eaggr.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.groupByKeys {
+		for j, groupBy := range oa.eaggr.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.groupByKeys {
+	for i, groupByKey := range oa.eaggr.GroupByKeys {
 		if referenced[i] {
 			continue
 		}
diff --git a/go/vt/vtgate/vcursor_impl.go b/go/vt/vtgate/vcursor_impl.go
index 9b5ffb5160..17ceee1909 100644
--- a/go/vt/vtgate/vcursor_impl.go
+++ b/go/vt/vtgate/vcursor_impl.go
@@ -43,7 +43,6 @@ import (
 	vschemapb "vitess.io/vitess/go/vt/proto/vschema"
 	vtgatepb "vitess.io/vitess/go/vt/proto/vtgate"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
-	"vitess.io/vitess/go/vt/schema"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/srvtopo"
 	"vitess.io/vitess/go/vt/topo"
@@ -638,19 +637,6 @@ func (vc *vcursorImpl) TabletType() topodatapb.TabletType {
 	return vc.tabletType
 }
 
-// SubmitOnlineDDL implements the VCursor interface
-func (vc *vcursorImpl) SubmitOnlineDDL(onlineDDl *schema.OnlineDDL) error {
-	if vc.topoServer == nil {
-		return vterrors.New(vtrpcpb.Code_INTERNAL, "Unable to apply DDL toposerver unavailable, ensure this vtgate is not using filtered keyspaces")
-	}
-	conn, err := vc.topoServer.ConnForCell(vc.ctx, topo.GlobalCell)
-	if err != nil {
-		return err
-	}
-	// Submit an online schema change by writing a migration request in topo
-	return onlineDDl.WriteTopo(vc.ctx, conn, schema.MigrationRequestsPath())
-}
-
 func commentedShardQueries(shardQueries []*querypb.BoundQuery, marginComments sqlparser.MarginComments) []*querypb.BoundQuery {
 	if marginComments.Leading == "" && marginComments.Trailing == "" {
 		return shardQueries


Test patch:
diff --git a/go/test/endtoend/cluster/vtctld_process.go b/go/test/endtoend/cluster/vtctld_process.go
index 64a5e2bed5..89bd754449 100644
--- a/go/test/endtoend/cluster/vtctld_process.go
+++ b/go/test/endtoend/cluster/vtctld_process.go
@@ -67,8 +67,6 @@ func (vtctld *VtctldProcess) Setup(cell string, extraArgs ...string) (err error)
 		// hard-code these two soon-to-be deprecated drain values.
 		"-wait_for_drain_sleep_rdonly", "1s",
 		"-wait_for_drain_sleep_replica", "1s",
-		// short online-ddl check interval to hasten tests
-		"-online_ddl_check_interval", "2s",
 		"-log_dir", vtctld.LogDir,
 		"-port", fmt.Sprintf("%d", vtctld.Port),
 		"-grpc_port", fmt.Sprintf("%d", vtctld.GrpcPort),
diff --git a/go/test/endtoend/onlineddl/ghost/onlineddl_ghost_test.go b/go/test/endtoend/onlineddl/ghost/onlineddl_ghost_test.go
index a1c6200b07..9c47e4e0d4 100644
--- a/go/test/endtoend/onlineddl/ghost/onlineddl_ghost_test.go
+++ b/go/test/endtoend/onlineddl/ghost/onlineddl_ghost_test.go
@@ -160,7 +160,6 @@ func TestMain(m *testing.M) {
 			"-schema_change_dir", schemaChangeDirectory,
 			"-schema_change_controller", "local",
 			"-schema_change_check_interval", "1",
-			"-online_ddl_check_interval", "3s",
 		}
 
 		clusterInstance.VtTabletExtraArgs = []string{
diff --git a/go/test/endtoend/onlineddl/revert/onlineddl_revert_test.go b/go/test/endtoend/onlineddl/revert/onlineddl_revert_test.go
index dccda3dcf5..3f5e1af498 100644
--- a/go/test/endtoend/onlineddl/revert/onlineddl_revert_test.go
+++ b/go/test/endtoend/onlineddl/revert/onlineddl_revert_test.go
@@ -182,7 +182,6 @@ func TestMain(m *testing.M) {
 			"-schema_change_dir", schemaChangeDirectory,
 			"-schema_change_controller", "local",
 			"-schema_change_check_interval", "1",
-			"-online_ddl_check_interval", "3s",
 		}
 
 		clusterInstance.VtTabletExtraArgs = []string{
diff --git a/go/test/endtoend/onlineddl/vrepl/onlineddl_vrepl_test.go b/go/test/endtoend/onlineddl/vrepl/onlineddl_vrepl_test.go
index a6f75f6c0a..6b3e572952 100644
--- a/go/test/endtoend/onlineddl/vrepl/onlineddl_vrepl_test.go
+++ b/go/test/endtoend/onlineddl/vrepl/onlineddl_vrepl_test.go
@@ -163,7 +163,6 @@ func TestMain(m *testing.M) {
 		clusterInstance.VtctldExtraArgs = []string{
 			"-schema_change_dir", schemaChangeDirectory,
 			"-schema_change_controller", "local",
-			"-online_ddl_check_interval", "3s",
 			"-schema_change_check_interval", "1",
 		}
 
diff --git a/go/test/endtoend/onlineddl/vrepl_stress/onlineddl_vrepl_mini_stress_test.go b/go/test/endtoend/onlineddl/vrepl_stress/onlineddl_vrepl_mini_stress_test.go
index 1559e1320d..93d25d4b5b 100644
--- a/go/test/endtoend/onlineddl/vrepl_stress/onlineddl_vrepl_mini_stress_test.go
+++ b/go/test/endtoend/onlineddl/vrepl_stress/onlineddl_vrepl_mini_stress_test.go
@@ -212,7 +212,6 @@ func TestMain(m *testing.M) {
 			"-schema_change_dir", schemaChangeDirectory,
 			"-schema_change_controller", "local",
 			"-schema_change_check_interval", "1",
-			"-online_ddl_check_interval", "3s",
 		}
 
 		clusterInstance.VtTabletExtraArgs = []string{
diff --git a/go/test/endtoend/onlineddl/vrepl_stress_suite/onlineddl_vrepl_stress_suite_test.go b/go/test/endtoend/onlineddl/vrepl_stress_suite/onlineddl_vrepl_stress_suite_test.go
index 502787d970..3578501512 100644
--- a/go/test/endtoend/onlineddl/vrepl_stress_suite/onlineddl_vrepl_stress_suite_test.go
+++ b/go/test/endtoend/onlineddl/vrepl_stress_suite/onlineddl_vrepl_stress_suite_test.go
@@ -409,7 +409,6 @@ func TestMain(m *testing.M) {
 			"-schema_change_dir", schemaChangeDirectory,
 			"-schema_change_controller", "local",
 			"-schema_change_check_interval", "1",
-			"-online_ddl_check_interval", "3s",
 		}
 
 		// -vstream_packet_size is set to a small value that ensures we get multiple stream iterations,
diff --git a/go/test/endtoend/onlineddl/vrepl_suite/onlineddl_vrepl_suite_test.go b/go/test/endtoend/onlineddl/vrepl_suite/onlineddl_vrepl_suite_test.go
index 2bdc462ac6..3f47510774 100644
--- a/go/test/endtoend/onlineddl/vrepl_suite/onlineddl_vrepl_suite_test.go
+++ b/go/test/endtoend/onlineddl/vrepl_suite/onlineddl_vrepl_suite_test.go
@@ -79,7 +79,6 @@ func TestMain(m *testing.M) {
 			"-schema_change_dir", schemaChangeDirectory,
 			"-schema_change_controller", "local",
 			"-schema_change_check_interval", "1",
-			"-online_ddl_check_interval", "2s",
 		}
 
 		clusterInstance.VtTabletExtraArgs = []string{
diff --git a/go/test/endtoend/schemadiff/vrepl/schemadiff_vrepl_suite_test.go b/go/test/endtoend/schemadiff/vrepl/schemadiff_vrepl_suite_test.go
index 39134ee755..1c64069a07 100644
--- a/go/test/endtoend/schemadiff/vrepl/schemadiff_vrepl_suite_test.go
+++ b/go/test/endtoend/schemadiff/vrepl/schemadiff_vrepl_suite_test.go
@@ -85,7 +85,6 @@ func TestMain(m *testing.M) {
 			"-schema_change_dir", schemaChangeDirectory,
 			"-schema_change_controller", "local",
 			"-schema_change_check_interval", "1",
-			"-online_ddl_check_interval", "2s",
 		}
 
 		clusterInstance.VtTabletExtraArgs = []string{
diff --git a/go/vt/schema/online_ddl_test.go b/go/vt/schema/online_ddl_test.go
index 00b7b2a094..686940fc3a 100644
--- a/go/vt/schema/online_ddl_test.go
+++ b/go/vt/schema/online_ddl_test.go
@@ -210,9 +210,6 @@ func TestNewOnlineDDL(t *testing.T) {
 		NewDDLStrategySetting(DDLStrategyVitess, ""),
 		NewDDLStrategySetting(DDLStrategyOnline, "-singleton"),
 	}
-	require.True(t, strategies[0].IsSkipTopo())
-	require.True(t, strategies[1].IsSkipTopo())
-	require.True(t, strategies[2].IsSkipTopo())
 
 	for _, ts := range tt {
 		t.Run(ts.sql, func(t *testing.T) {
@@ -224,7 +221,6 @@ func TestNewOnlineDDL(t *testing.T) {
 						return
 					}
 					assert.NoError(t, err)
-					require.True(t, stgy.IsSkipTopo(), "IsSkipTopo() should always be true")
 					// onlineDDL.SQL enriched with /*vt+ ... */ comment
 					assert.Contains(t, onlineDDL.SQL, hex.EncodeToString([]byte(onlineDDL.UUID)))
 					assert.Contains(t, onlineDDL.SQL, hex.EncodeToString([]byte(migrationContext)))
diff --git a/go/vt/vtgate/engine/fake_vcursor_test.go b/go/vt/vtgate/engine/fake_vcursor_test.go
index 9892c182e5..c268f99121 100644
--- a/go/vt/vtgate/engine/fake_vcursor_test.go
+++ b/go/vt/vtgate/engine/fake_vcursor_test.go
@@ -33,7 +33,6 @@ import (
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/test/utils"
 	"vitess.io/vitess/go/vt/key"
-	"vitess.io/vitess/go/vt/schema"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/srvtopo"
 	"vitess.io/vitess/go/vt/vtgate/vindexes"
@@ -276,10 +275,6 @@ func (t *noopVCursor) ResolveDestinationsMultiCol(keyspace string, ids [][]sqlty
 	panic("unimplemented")
 }
 
-func (t *noopVCursor) SubmitOnlineDDL(onlineDDl *schema.OnlineDDL) error {
-	panic("unimplemented")
-}
-
 func (t *noopVCursor) GetDBDDLPluginName() string {
 	panic("unimplemented")
 }
@@ -449,11 +444,6 @@ func (f *loggingVCursor) AutocommitApproval() bool {
 	return true
 }
 
-func (f *loggingVCursor) SubmitOnlineDDL(onlineDDL *schema.OnlineDDL) error {
-	f.log = append(f.log, fmt.Sprintf("SubmitOnlineDDL: %s", onlineDDL.ToString()))
-	return nil
-}
-
 func (f *loggingVCursor) ExecuteStandalone(query string, bindvars map[string]*querypb.BindVariable, rs *srvtopo.ResolvedShard) (*sqltypes.Result, error) {
 	f.log = append(f.log, fmt.Sprintf("ExecuteStandalone %s %v %s %s", query, printBindVars(bindvars), rs.Target.Keyspace, rs.Target.Shard))
 	return f.nextResult()
diff --git a/go/vt/vtgate/engine/ordered_aggregate_test.go b/go/vt/vtgate/engine/ordered_aggregate_test.go
index e8fcc0cac4..3de0ca0f9e 100644
--- a/go/vt/vtgate/engine/ordered_aggregate_test.go
+++ b/go/vt/vtgate/engine/ordered_aggregate_test.go
@@ -656,17 +656,94 @@ func TestMerge(t *testing.T) {
 		"1|3|2.8|2|bc",
 	)
 
-	merged, _, err := merge(fields, r.Rows[0], r.Rows[1], nil, nil, oa.Aggregates)
+	merged, _, err := oa.merge(fields, r.Rows[0], r.Rows[1], nil, nil)
 	assert.NoError(err)
 	want := sqltypes.MakeTestResult(fields, "1|5|6.0|2|bc").Rows[0]
 	assert.Equal(want, merged)
 
 	// swap and retry
-	merged, _, err = merge(fields, r.Rows[1], r.Rows[0], nil, nil, oa.Aggregates)
+	merged, _, err = oa.merge(fields, r.Rows[1], r.Rows[0], nil, nil)
 	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{
@@ -679,6 +756,7 @@ func TestOrderedAggregateExecuteGtid(t *testing.T) {
 		Shard:    "80-",
 		Gtid:     "b",
 	})
+	fmt.Println(vgtid.String())
 
 	fp := &fakePrimitive{
 		results: []*sqltypes.Result{sqltypes.MakeTestResult(
@@ -1001,7 +1079,7 @@ func TestOrderedAggregateCollate(t *testing.T) {
 	collationID, _ := collations.Local().LookupID("utf8mb4_0900_ai_ci")
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateSum,
+			Opcode: AggregateCount,
 			Col:    1,
 		}},
 		GroupByKeys: []*GroupByParams{{KeyCol: 0, CollationID: collationID}},
@@ -1044,7 +1122,7 @@ func TestOrderedAggregateCollateAS(t *testing.T) {
 	collationID, _ := collations.Local().LookupID("utf8mb4_0900_as_ci")
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateSum,
+			Opcode: AggregateCount,
 			Col:    1,
 		}},
 		GroupByKeys: []*GroupByParams{{KeyCol: 0, CollationID: collationID}},
@@ -1089,7 +1167,7 @@ func TestOrderedAggregateCollateKS(t *testing.T) {
 	collationID, _ := collations.Local().LookupID("utf8mb4_ja_0900_as_cs_ks")
 	oa := &OrderedAggregate{
 		Aggregates: []*AggregateParams{{
-			Opcode: AggregateSum,
+			Opcode: AggregateCount,
 			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
deleted file mode 100644
index 0e8987e4d4..0000000000
--- a/go/vt/vtgate/engine/scalar_aggregation_test.go
+++ /dev/null
@@ -1,135 +0,0 @@
-/*
-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/planbuilder/testdata/aggr_cases.txt b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
index 797dea2683..2677992171 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": "Scalar",
+    "Variant": "Ordered",
     "Aggregates": "count(0)",
     "Inputs": [
       {
@@ -302,7 +302,7 @@ Gen4 plan same as above
   "Original": "select count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "Aggregates": "sum(0)",
     "Inputs": [
       {
@@ -349,7 +349,7 @@ Gen4 plan same as above
   "Original": "select sum(col) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "Aggregates": "min(0)",
     "Inputs": [
       {
@@ -396,7 +396,7 @@ Gen4 plan same as above
   "Original": "select min(col) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "Aggregates": "max(0)",
     "Inputs": [
       {
@@ -443,7 +443,7 @@ Gen4 plan same as above
   "Original": "select max(col) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Scalar",
+    "Variant": "Ordered",
     "Aggregates": "max(0) AS max(col)",
     "Inputs": [
       {
@@ -870,7 +870,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0)",
         "Inputs": [
           {
@@ -900,7 +900,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "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": "Scalar",
+    "Variant": "Ordered",
     "Aggregates": "count(1)",
     "Inputs": [
       {
@@ -1130,7 +1130,7 @@ Gen4 plan same as above
   "Original": "select count(distinct col2) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "Aggregates": "count(0) AS count(*)",
     "Inputs": [
       {
@@ -2214,7 +2214,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0)",
         "Inputs": [
           {
@@ -2238,7 +2238,7 @@ Gen4 plan same as above
   "Original": "select distinct count(*) from user",
   "Instructions": {
     "OperatorType": "Aggregate",
-    "Variant": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "Aggregates": "count(0) AS count(*)",
     "Inputs": [
       {
@@ -2743,7 +2743,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2775,7 +2775,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2807,7 +2807,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2839,7 +2839,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2871,7 +2871,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2903,7 +2903,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2935,7 +2935,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0) AS a",
         "Inputs": [
           {
@@ -2967,7 +2967,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "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 03fc842fce..a2e917d0b1 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": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0)",
         "Inputs": [
           {
@@ -3730,7 +3730,7 @@ Gen4 plan same as above
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "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 c8563f514b..94a973c370 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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "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 003543cf68..88abec9aae 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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+        "Variant": "Ordered",
         "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": "Scalar",
+        "Variant": "Ordered",
         "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 55ce84405e..17f921e3fc 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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "Aggregates": "count(0) AS count(*)",
     "Inputs": [
       {
@@ -1766,7 +1766,7 @@ Gen4 plan same as above
       },
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0)",
         "Inputs": [
           {
@@ -1810,7 +1810,7 @@ Gen4 plan same as above
       },
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0) AS count(*)",
         "Inputs": [
           {
@@ -1902,7 +1902,7 @@ Gen4 plan same as above
       },
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "count(0)",
         "Inputs": [
           {
@@ -1948,7 +1948,7 @@ Gen4 plan same as above
       },
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "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 2c8971bb99..8266e3e553 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": "Scalar",
+    "Variant": "Ordered",
     "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": "Scalar",
+    "Variant": "Ordered",
     "Aggregates": "sum(0) AS revenue",
     "Inputs": [
       {
@@ -129,7 +129,7 @@ Gen4 error: unsupported: in scatter query: complex aggregate expression
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "Aggregates": "max(0)",
         "Inputs": [
           {
@@ -200,7 +200,7 @@ Gen4 error: unsupported: in scatter query: complex aggregate expression
     "Inputs": [
       {
         "OperatorType": "Aggregate",
-        "Variant": "Scalar",
+        "Variant": "Ordered",
         "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": "Scalar",
+    "Variant": "Ordered",
     "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 57d8c6f504..3ab3ff0e67 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": "Scalar",
+            "Variant": "Ordered",
             "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": "Scalar",
+            "Variant": "Ordered",
             "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": "Scalar",
+            "Variant": "Ordered",
             "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": "Scalar",
+            "Variant": "Ordered",
             "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 37bf043858..575cc25bd8 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": "Scalar",
+    "Variant": "Ordered",
     "Aggregates": "count(1)",
     "Inputs": [
       {