Repository: vitessio/vitess
Pull request #11221 (2022-09-14 17:45:21)

addressing review comments from #11088
<!--
  Thank you for your contribution to the Vitess project.
  How to contribute: https://vitess.io/docs/contributing/
  Please first make sure there is an open Issue to discuss the feature/fix suggested in this PR.
  If this is a new feature, please mark the Issue as "RFC".
 -->

<!-- if this PR is Work in Progress please create it as a Draft Pull Request -->

## 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. -->

This PR addressed the additional review comments from the PR https://github.com/vitessio/vitess/pull/11088

## Related Issue(s)

<!-- List related issues and pull requests. If this PR fixes an issue, please add it using Fixes #????  -->
- #9706 

## Checklist

-   [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/doc/releasenotes/15_0_0_summary.md b/doc/releasenotes/15_0_0_summary.md
index 7054f0eb77..eec611b9ab 100644
--- a/doc/releasenotes/15_0_0_summary.md
+++ b/doc/releasenotes/15_0_0_summary.md
@@ -219,6 +219,18 @@ $ curl -s http://127.0.0.1:15100/debug/vars | jq . | grep Throttler
 
 ### Mysql Compatibility
 
+#### System Settings
+Vitess supported system settings from release 7.0 onwards, but it was always with a pinch of salt.
+As soon as a client session changes a default system setting, the mysql connection gets blocked for it.
+This leads to clients running out of mysql connections. 
+The clients were instructed to use this to a minimum and try to set those changed system settings as default on the mysql.
+
+With this release, Vitess can handle system settings changes in a much better way and the clients can use it more freely.
+Vitess now pools those changed settings and does not reserve it for any particular session. 
+
+This feature can be enabled by setting `queryserver-enable-settings-pool` flag on the vttablet. It is disabled by default.
+In future releases, we will make this flag enabled by default.
+
 #### Lookup Vindexes
 
 Lookup vindexes now support a new parameter `multi_shard_autocommit`. If this is set to `true`, lookup vindex dml queries will be sent as autocommit to all shards instead of being wrapped in a transaction.
diff --git a/go/flags/endtoend/vttablet.txt b/go/flags/endtoend/vttablet.txt
index 0c68b5b3ff..429c404407 100644
--- a/go/flags/endtoend/vttablet.txt
+++ b/go/flags/endtoend/vttablet.txt
@@ -341,8 +341,8 @@ Usage of vttablet:
       --queryserver-config-txpool-timeout float                          query server transaction pool timeout, it is how long vttablet waits if tx pool is full (default 1)
       --queryserver-config-txpool-waiter-cap int                         query server transaction pool waiter limit, this is the maximum number of transactions that can be queued waiting to get a connection (default 5000)
       --queryserver-config-warn-result-size int                          query server result size warning threshold, warn if number of rows returned from vttablet for non-streaming queries exceeds this
+      --queryserver-enable-settings-pool                                 Enable pooling of connections with modified system settings
       --queryserver_enable_online_ddl                                    Enable online DDL. (default true)
-      --queryserver_enable_settings_pool                                 Enable pooling of connections with modified system settings
       --redact-debug-ui-queries                                          redact full queries and bind variables from debug UI
       --relay_log_max_items int                                          Maximum number of rows for VReplication target buffering. (default 5000)
       --relay_log_max_size int                                           Maximum buffer size (in bytes) for VReplication target buffering. If single rows are larger than this, a single row is buffered at a time. (default 250000)
diff --git a/go/vt/vtgate/engine/delete.go b/go/vt/vtgate/engine/delete.go
index 4a8257741c..d748b572b6 100644
--- a/go/vt/vtgate/engine/delete.go
+++ b/go/vt/vtgate/engine/delete.go
@@ -62,7 +62,7 @@ func (del *Delete) TryExecute(ctx context.Context, vcursor VCursor, bindVars map
 	switch del.Opcode {
 	case Unsharded:
 		return del.execUnsharded(ctx, vcursor, bindVars, rss)
-	case Equal, IN, Scatter, ByDestination, SubShard:
+	case Equal, IN, Scatter, ByDestination, SubShard, EqualUnique:
 		return del.execMultiDestination(ctx, vcursor, bindVars, rss, del.deleteVindexEntries)
 	default:
 		// Unreachable.
diff --git a/go/vt/vtgate/planbuilder/abstract/delete.go b/go/vt/vtgate/planbuilder/abstract/delete.go
new file mode 100644
index 0000000000..8aa9445d64
--- /dev/null
+++ b/go/vt/vtgate/planbuilder/abstract/delete.go
@@ -0,0 +1,60 @@
+/*
+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 abstract
+
+import (
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/vterrors"
+	"vitess.io/vitess/go/vt/vtgate/semantics"
+)
+
+type Delete struct {
+	Table     *QueryTable
+	TableInfo semantics.TableInfo
+	AST       *sqlparser.Delete
+}
+
+var _ LogicalOperator = (*Delete)(nil)
+
+// TableID implements the LogicalOperator interface
+func (d *Delete) TableID() semantics.TableSet {
+	return d.Table.ID
+}
+
+// UnsolvedPredicates implements the LogicalOperator interface
+func (d *Delete) UnsolvedPredicates(semTable *semantics.SemTable) []sqlparser.Expr {
+	return nil
+}
+
+// CheckValid implements the LogicalOperator interface
+func (d *Delete) CheckValid() error {
+	return nil
+}
+
+// iLogical implements the LogicalOperator interface
+func (d *Delete) iLogical() {}
+
+// PushPredicate implements the LogicalOperator interface
+func (d *Delete) PushPredicate(expr sqlparser.Expr, semTable *semantics.SemTable) (LogicalOperator, error) {
+	return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "can't accept predicates")
+}
+
+// Compact implements the LogicalOperator interface
+func (d *Delete) Compact(semTable *semantics.SemTable) (LogicalOperator, error) {
+	return d, nil
+}
diff --git a/go/vt/vtgate/planbuilder/abstract/operator.go b/go/vt/vtgate/planbuilder/abstract/operator.go
index ffaa779e77..8d36dc1c2d 100644
--- a/go/vt/vtgate/planbuilder/abstract/operator.go
+++ b/go/vt/vtgate/planbuilder/abstract/operator.go
@@ -174,6 +174,8 @@ func CreateLogicalOperatorFromAST(selStmt sqlparser.Statement, semTable *semanti
 		op, err = createOperatorFromUnion(node, semTable)
 	case *sqlparser.Update:
 		op, err = createOperatorFromUpdate(node, semTable)
+	case *sqlparser.Delete:
+		op, err = createOperatorFromDelete(node, semTable)
 	default:
 		err = vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "%T: operator not yet supported", selStmt)
 	}
@@ -234,32 +236,57 @@ func createOperatorFromSelect(sel *sqlparser.Select, semTable *semantics.SemTabl
 }
 
 func createOperatorFromUpdate(updStmt *sqlparser.Update, semTable *semantics.SemTable) (LogicalOperator, error) {
+	tableInfo, qt, err := createQueryTableForDML(updStmt.TableExprs[0], semTable, updStmt.Where)
+	if err != nil {
+		return nil, err
+	}
+
+	assignments := make(map[string]sqlparser.Expr)
+	for _, set := range updStmt.Exprs {
+		assignments[set.Name.Name.String()] = set.Expr
+	}
+
+	u := &Update{
+		Table:       qt,
+		Assignments: assignments,
+		AST:         updStmt,
+		TableInfo:   tableInfo,
+	}
+
 	subq, err := createSubqueryFromStatement(updStmt, semTable)
 	if err != nil {
 		return nil, err
 	}
-	alTbl, ok := updStmt.TableExprs[0].(*sqlparser.AliasedTableExpr)
+	if subq == nil {
+		return u, nil
+	}
+	subq.Outer = u
+	return subq, nil
+}
+
+func createQueryTableForDML(tableExpr sqlparser.TableExpr, semTable *semantics.SemTable, whereClause *sqlparser.Where) (semantics.TableInfo, *QueryTable, error) {
+	alTbl, ok := tableExpr.(*sqlparser.AliasedTableExpr)
 	if !ok {
-		return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "expected AliasedTableExpr")
+		return nil, nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "expected AliasedTableExpr")
 	}
 	tblName, ok := alTbl.Expr.(sqlparser.TableName)
 	if !ok {
-		return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "expected TableName")
+		return nil, nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "expected TableName")
 	}
 
 	tableID := semTable.TableSetFor(alTbl)
 	tableInfo, err := semTable.TableInfoFor(tableID)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	if tableInfo.IsInfSchema() {
-		return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "can't update information schema tables")
+		return nil, nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "can't update information schema tables")
 	}
 
 	var predicates []sqlparser.Expr
-	if updStmt.Where != nil {
-		predicates = sqlparser.SplitAndExpression(nil, updStmt.Where.Expr)
+	if whereClause != nil {
+		predicates = sqlparser.SplitAndExpression(nil, whereClause.Expr)
 	}
 	qt := &QueryTable{
 		ID:          tableID,
@@ -268,17 +295,24 @@ func createOperatorFromUpdate(updStmt *sqlparser.Update, semTable *semantics.Sem
 		Predicates:  predicates,
 		IsInfSchema: false,
 	}
+	return tableInfo, qt, nil
+}
 
-	assignments := make(map[string]sqlparser.Expr)
-	for _, set := range updStmt.Exprs {
-		assignments[set.Name.Name.String()] = set.Expr
+func createOperatorFromDelete(deleteStmt *sqlparser.Delete, semTable *semantics.SemTable) (LogicalOperator, error) {
+	tableInfo, qt, err := createQueryTableForDML(deleteStmt.TableExprs[0], semTable, deleteStmt.Where)
+	if err != nil {
+		return nil, err
 	}
 
-	u := &Update{
-		Table:       qt,
-		Assignments: assignments,
-		AST:         updStmt,
-		TableInfo:   tableInfo,
+	u := &Delete{
+		Table:     qt,
+		AST:       deleteStmt,
+		TableInfo: tableInfo,
+	}
+
+	subq, err := createSubqueryFromStatement(deleteStmt, semTable)
+	if err != nil {
+		return nil, err
 	}
 	if subq == nil {
 		return u, nil
diff --git a/go/vt/vtgate/planbuilder/builder.go b/go/vt/vtgate/planbuilder/builder.go
index 182de3e201..ce299ae0f3 100644
--- a/go/vt/vtgate/planbuilder/builder.go
+++ b/go/vt/vtgate/planbuilder/builder.go
@@ -221,7 +221,11 @@ func createInstructionFor(query string, stmt sqlparser.Statement, reservedVars *
 		}
 		return buildRoutePlan(stmt, reservedVars, vschema, configuredPlanner)
 	case *sqlparser.Delete:
-		return buildRoutePlan(stmt, reservedVars, vschema, buildDeletePlan)
+		configuredPlanner, err := getConfiguredPlanner(vschema, buildDeletePlan, stmt, query)
+		if err != nil {
+			return nil, err
+		}
+		return buildRoutePlan(stmt, reservedVars, vschema, configuredPlanner)
 	case *sqlparser.Union:
 		configuredPlanner, err := getConfiguredPlanner(vschema, buildUnionPlan, stmt, query)
 		if err != nil {
diff --git a/go/vt/vtgate/planbuilder/delete.go b/go/vt/vtgate/planbuilder/delete.go
index f446b9a490..f93299cf79 100644
--- a/go/vt/vtgate/planbuilder/delete.go
+++ b/go/vt/vtgate/planbuilder/delete.go
@@ -25,51 +25,53 @@ import (
 )
 
 // buildDeletePlan builds the instructions for a DELETE statement.
-func buildDeletePlan(stmt sqlparser.Statement, reservedVars *sqlparser.ReservedVars, vschema plancontext.VSchema) (*planResult, error) {
-	del := stmt.(*sqlparser.Delete)
-	if del.With != nil {
-		return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: with expression in delete statement")
-	}
-	var err error
-	if len(del.TableExprs) == 1 && len(del.Targets) == 1 {
-		del, err = rewriteSingleTbl(del)
+func buildDeletePlan(string) stmtPlanner {
+	return func(stmt sqlparser.Statement, reservedVars *sqlparser.ReservedVars, vschema plancontext.VSchema) (*planResult, error) {
+		del := stmt.(*sqlparser.Delete)
+		if del.With != nil {
+			return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: with expression in delete statement")
+		}
+		var err error
+		if len(del.TableExprs) == 1 && len(del.Targets) == 1 {
+			del, err = rewriteSingleTbl(del)
+			if err != nil {
+				return nil, err
+			}
+		}
+		dml, tables, ksidVindex, err := buildDMLPlan(vschema, "delete", del, reservedVars, del.TableExprs, del.Where, del.OrderBy, del.Limit, del.Comments, del.Targets)
 		if err != nil {
 			return nil, err
 		}
-	}
-	dml, tables, ksidVindex, err := buildDMLPlan(vschema, "delete", del, reservedVars, del.TableExprs, del.Where, del.OrderBy, del.Limit, del.Comments, del.Targets)
-	if err != nil {
-		return nil, err
-	}
-	edel := &engine.Delete{DML: dml}
-	if dml.Opcode == engine.Unsharded {
-		return newPlanResult(edel, tables...), nil
-	}
+		edel := &engine.Delete{DML: dml}
+		if dml.Opcode == engine.Unsharded {
+			return newPlanResult(edel, tables...), nil
+		}
 
-	if len(del.Targets) > 1 {
-		return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "multi-table delete statement in not supported in sharded database")
-	}
+		if len(del.Targets) > 1 {
+			return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "multi-table delete statement in not supported in sharded database")
+		}
 
-	edelTable, err := edel.GetSingleTable()
-	if err != nil {
-		return nil, err
-	}
-	if len(del.Targets) == 1 && del.Targets[0].Name != edelTable.Name {
-		return nil, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.UnknownTable, "Unknown table '%s' in MULTI DELETE", del.Targets[0].Name.String())
-	}
+		edelTable, err := edel.GetSingleTable()
+		if err != nil {
+			return nil, err
+		}
+		if len(del.Targets) == 1 && del.Targets[0].Name != edelTable.Name {
+			return nil, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.UnknownTable, "Unknown table '%s' in MULTI DELETE", del.Targets[0].Name.String())
+		}
 
-	if len(edelTable.Owned) > 0 {
-		aTblExpr, ok := del.TableExprs[0].(*sqlparser.AliasedTableExpr)
-		if !ok {
-			return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: delete on complex table expression")
+		if len(edelTable.Owned) > 0 {
+			aTblExpr, ok := del.TableExprs[0].(*sqlparser.AliasedTableExpr)
+			if !ok {
+				return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: delete on complex table expression")
+			}
+			tblExpr := &sqlparser.AliasedTableExpr{Expr: sqlparser.TableName{Name: edelTable.Name}, As: aTblExpr.As}
+			edel.OwnedVindexQuery = generateDMLSubquery(tblExpr, del.Where, del.OrderBy, del.Limit, edelTable, ksidVindex.Columns)
+			edel.KsidVindex = ksidVindex.Vindex
+			edel.KsidLength = len(ksidVindex.Columns)
 		}
-		tblExpr := &sqlparser.AliasedTableExpr{Expr: sqlparser.TableName{Name: edelTable.Name}, As: aTblExpr.As}
-		edel.OwnedVindexQuery = generateDMLSubquery(tblExpr, del.Where, del.OrderBy, del.Limit, edelTable, ksidVindex.Columns)
-		edel.KsidVindex = ksidVindex.Vindex
-		edel.KsidLength = len(ksidVindex.Columns)
-	}
 
-	return newPlanResult(edel, tables...), nil
+		return newPlanResult(edel, tables...), nil
+	}
 }
 
 func rewriteSingleTbl(del *sqlparser.Delete) (*sqlparser.Delete, error) {
diff --git a/go/vt/vtgate/planbuilder/gen4_planner.go b/go/vt/vtgate/planbuilder/gen4_planner.go
index deba9b87e7..0dfcee1b75 100644
--- a/go/vt/vtgate/planbuilder/gen4_planner.go
+++ b/go/vt/vtgate/planbuilder/gen4_planner.go
@@ -17,6 +17,8 @@ limitations under the License.
 package planbuilder
 
 import (
+	"errors"
+
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/sqlparser"
@@ -38,6 +40,8 @@ func gen4Planner(query string, plannerVersion querypb.ExecuteOptions_PlannerVers
 			return gen4SelectStmtPlanner(query, plannerVersion, stmt, reservedVars, vschema)
 		case *sqlparser.Update:
 			return gen4UpdateStmtPlanner(plannerVersion, stmt, reservedVars, vschema)
+		case *sqlparser.Delete:
+			return gen4DeleteStmtPlanner(plannerVersion, stmt, reservedVars, vschema)
 		default:
 			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "%T not yet supported", stmt)
 		}
@@ -322,9 +326,97 @@ func gen4UpdateStmtPlanner(
 	return newPlanResult(plan.Primitive(), tablesFromSemantics(semTable)...), nil
 }
 
-func rewriteRoutedTables(updStmt *sqlparser.Update, vschema plancontext.VSchema) (err error) {
+func gen4DeleteStmtPlanner(
+	version querypb.ExecuteOptions_PlannerVersion,
+	deleteStmt *sqlparser.Delete,
+	reservedVars *sqlparser.ReservedVars,
+	vschema plancontext.VSchema,
+) (*planResult, error) {
+	if deleteStmt.With != nil {
+		return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: with expression in delete statement")
+	}
+
+	var err error
+	if len(deleteStmt.TableExprs) == 1 && len(deleteStmt.Targets) == 1 {
+		deleteStmt, err = rewriteSingleTbl(deleteStmt)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	ksName := ""
+	if ks, _ := vschema.DefaultKeyspace(); ks != nil {
+		ksName = ks.Name
+	}
+	semTable, err := semantics.Analyze(deleteStmt, ksName, vschema)
+	if err != nil {
+		return nil, err
+	}
+
+	// record any warning as planner warning.
+	vschema.PlannerWarning(semTable.Warning)
+	err = rewriteRoutedTables(deleteStmt, vschema)
+	if err != nil {
+		return nil, err
+	}
+
+	if ks, tables := semTable.SingleUnshardedKeyspace(); ks != nil {
+		edml := engine.NewDML()
+		edml.Keyspace = ks
+		edml.Table = tables
+		edml.Opcode = engine.Unsharded
+		edml.Query = generateQuery(deleteStmt)
+		del := &engine.Delete{DML: edml}
+		return newPlanResult(del, tablesFromSemantics(semTable)...), nil
+	}
+
+	if err := checkIfDeleteSupported(deleteStmt, semTable); err != nil {
+		return nil, err
+	}
+
+	err = queryRewrite(semTable, reservedVars, deleteStmt)
+	if err != nil {
+		return nil, err
+	}
+
+	logical, err := abstract.CreateLogicalOperatorFromAST(deleteStmt, semTable)
+	if err != nil {
+		return nil, err
+	}
+	err = logical.CheckValid()
+	if err != nil {
+		return nil, err
+	}
+
+	ctx := plancontext.NewPlanningContext(reservedVars, semTable, vschema, version)
+
+	physOp, err := physical.CreatePhysicalOperator(ctx, logical)
+	if err != nil {
+		return nil, err
+	}
+
+	plan, err := transformToLogicalPlan(ctx, physOp, true)
+	if err != nil {
+		return nil, err
+	}
+
+	plan, err = pushCommentDirectivesOnPlan(plan, deleteStmt)
+	if err != nil {
+		return nil, err
+	}
+
+	setLockOnAllSelect(plan)
+
+	if err := plan.WireupGen4(ctx); err != nil {
+		return nil, err
+	}
+
+	return newPlanResult(plan.Primitive(), tablesFromSemantics(semTable)...), nil
+}
+
+func rewriteRoutedTables(stmt sqlparser.Statement, vschema plancontext.VSchema) (err error) {
 	// Rewrite routed tables
-	_ = sqlparser.Rewrite(updStmt, func(cursor *sqlparser.Cursor) bool {
+	_ = sqlparser.Rewrite(stmt, func(cursor *sqlparser.Cursor) bool {
 		aliasTbl, isAlias := cursor.Node().(*sqlparser.AliasedTableExpr)
 		if !isAlias {
 			return err == nil
@@ -468,3 +560,54 @@ func pushCommentDirectivesOnPlan(plan logicalPlan, stmt sqlparser.Statement) (lo
 
 	return plan, nil
 }
+
+// checkIfDeleteSupported checks if the delete query is supported or we must return an error.
+func checkIfDeleteSupported(del *sqlparser.Delete, semTable *semantics.SemTable) error {
+	if semTable.NotUnshardedErr != nil {
+		return semTable.NotUnshardedErr
+	}
+
+	// Delete is only supported for a single TableExpr which is supposed to be an aliased expression
+	multiShardErr := errors.New("unsupported: multi-shard or vindex write statement")
+	if len(del.TableExprs) != 1 {
+		return multiShardErr
+	}
+	aliasedTableExpr, isAliasedExpr := del.TableExprs[0].(*sqlparser.AliasedTableExpr)
+	if !isAliasedExpr {
+		return multiShardErr
+	}
+
+	if len(del.Targets) > 1 {
+		return vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "multi-table delete statement in not supported in sharded database")
+	}
+
+	// Get the table information and the vindex table from it
+	ti, err := semTable.TableInfoFor(semTable.TableSetFor(aliasedTableExpr))
+	if err != nil {
+		return err
+	}
+	isSharded := false
+	vt := ti.GetVindexTable()
+	if vt != nil && vt.Keyspace != nil {
+		isSharded = vt.Keyspace.Sharded
+	}
+
+	err = sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
+		switch node.(type) {
+		case *sqlparser.Subquery, *sqlparser.DerivedTable:
+			// We have a subquery, so we must fail the planning.
+			// If this subquery and the table expression were all belonging to the same unsharded keyspace,
+			// we would have already created a plan for them before doing these checks.
+			if isSharded {
+				return false, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: subqueries in sharded DML")
+			}
+			return false, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: sharded subqueries in DML")
+		}
+		return true, nil
+	}, del)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
diff --git a/go/vt/vtgate/planbuilder/operator_transformers.go b/go/vt/vtgate/planbuilder/operator_transformers.go
index 9bc0fd986b..efffbd60f9 100644
--- a/go/vt/vtgate/planbuilder/operator_transformers.go
+++ b/go/vt/vtgate/planbuilder/operator_transformers.go
@@ -134,9 +134,11 @@ func transformApplyJoinPlan(ctx *plancontext.PlanningContext, n *physical.ApplyJ
 }
 
 func transformRoutePlan(ctx *plancontext.PlanningContext, op *physical.Route) (logicalPlan, error) {
-	upd, isUpdate := op.Source.(*physical.Update)
-	if isUpdate {
-		return transformUpdatePlan(ctx, op, upd)
+	switch src := op.Source.(type) {
+	case *physical.Update:
+		return transformUpdatePlan(ctx, op, src)
+	case *physical.Delete:
+		return transformDeletePlan(ctx, op, src)
 	}
 	tableNames, err := getAllTableNames(op)
 	if err != nil {
@@ -214,6 +216,48 @@ func transformUpdatePlan(ctx *plancontext.PlanningContext, op *physical.Route, u
 	return &primitiveWrapper{prim: e}, nil
 }
 
+func transformDeletePlan(ctx *plancontext.PlanningContext, op *physical.Route, del *physical.Delete) (logicalPlan, error) {
+	var vindex vindexes.Vindex
+	var values []evalengine.Expr
+	if op.Selected != nil {
+		vindex = op.Selected.FoundVindex
+		values = op.Selected.Values
+	}
+	ast := del.AST
+	replaceSubQuery(ctx, ast)
+	edml := &engine.DML{
+		Query: generateQuery(ast),
+		Table: []*vindexes.Table{
+			del.VTable,
+		},
+		OwnedVindexQuery: del.OwnedVindexQuery,
+		RoutingParameters: &engine.RoutingParameters{
+			Opcode:            op.RouteOpCode,
+			Keyspace:          op.Keyspace,
+			Vindex:            vindex,
+			Values:            values,
+			TargetDestination: op.TargetDestination,
+		},
+	}
+
+	directives := del.AST.GetParsedComments().Directives()
+	if directives.IsSet(sqlparser.DirectiveMultiShardAutocommit) {
+		edml.MultiShardAutocommit = true
+	}
+	edml.QueryTimeout = queryTimeout(directives)
+
+	e := &engine.Delete{}
+	e.DML = edml
+
+	if op.RouteOpCode != engine.Unsharded && del.OwnedVindexQuery != "" {
+		primary := del.VTable.ColumnVindexes[0]
+		e.DML.KsidVindex = primary.Vindex
+		e.DML.KsidLength = len(primary.Columns)
+	}
+
+	return &primitiveWrapper{prim: e}, nil
+}
+
 func replaceSubQuery(ctx *plancontext.PlanningContext, sel sqlparser.Statement) {
 	extractedSubqueries := ctx.SemTable.GetSubqueryNeedingRewrite()
 	if len(extractedSubqueries) == 0 {
diff --git a/go/vt/vtgate/planbuilder/physical/delete.go b/go/vt/vtgate/planbuilder/physical/delete.go
new file mode 100644
index 0000000000..23cb6f3372
--- /dev/null
+++ b/go/vt/vtgate/planbuilder/physical/delete.go
@@ -0,0 +1,77 @@
+/*
+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 physical
+
+import (
+	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/vtgate/planbuilder/abstract"
+	"vitess.io/vitess/go/vt/vtgate/semantics"
+	"vitess.io/vitess/go/vt/vtgate/vindexes"
+)
+
+type Delete struct {
+	QTable           *abstract.QueryTable
+	VTable           *vindexes.Table
+	OwnedVindexQuery string
+	AST              *sqlparser.Delete
+}
+
+var _ abstract.PhysicalOperator = (*Delete)(nil)
+var _ abstract.IntroducesTable = (*Delete)(nil)
+
+// TableID implements the PhysicalOperator interface
+func (d *Delete) TableID() semantics.TableSet {
+	return d.QTable.ID
+}
+
+// UnsolvedPredicates implements the PhysicalOperator interface
+func (d *Delete) UnsolvedPredicates(semTable *semantics.SemTable) []sqlparser.Expr {
+	return nil
+}
+
+// CheckValid implements the PhysicalOperator interface
+func (d *Delete) CheckValid() error {
+	return nil
+}
+
+// IPhysical implements the PhysicalOperator interface
+func (d *Delete) IPhysical() {}
+
+// Cost implements the PhysicalOperator interface
+func (d *Delete) Cost() int {
+	return 1
+}
+
+// Clone implements the PhysicalOperator interface
+func (d *Delete) Clone() abstract.PhysicalOperator {
+	return &Delete{
+		QTable:           d.QTable,
+		VTable:           d.VTable,
+		OwnedVindexQuery: d.OwnedVindexQuery,
+		AST:              d.AST,
+	}
+}
+
+// GetQTable implements the IntroducesTable interface
+func (d *Delete) GetQTable() *abstract.QueryTable {
+	return d.QTable
+}
+
+// GetVTable implements the IntroducesTable interface
+func (d *Delete) GetVTable() *vindexes.Table {
+	return d.VTable
+}
diff --git a/go/vt/vtgate/planbuilder/physical/route_planning.go b/go/vt/vtgate/planbuilder/physical/route_planning.go
index 5b52a89a00..0d67e5a974 100644
--- a/go/vt/vtgate/planbuilder/physical/route_planning.go
+++ b/go/vt/vtgate/planbuilder/physical/route_planning.go
@@ -21,12 +21,8 @@ import (
 	"fmt"
 	"io"
 
-	"vitess.io/vitess/go/vt/key"
-	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
-
-	querypb "vitess.io/vitess/go/vt/proto/query"
-
 	"vitess.io/vitess/go/mysql/collations"
+	"vitess.io/vitess/go/vt/key"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vtgate/engine"
@@ -36,6 +32,8 @@ import (
 	"vitess.io/vitess/go/vt/vtgate/semantics"
 	"vitess.io/vitess/go/vt/vtgate/vindexes"
 
+	querypb "vitess.io/vitess/go/vt/proto/query"
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 )
 
@@ -104,67 +102,142 @@ func CreatePhysicalOperator(ctx *plancontext.PlanningContext, opTree abstract.Lo
 
 		return filter, nil
 	case *abstract.Update:
-		var typ topodatapb.TabletType
-		var dest key.Destination
+		return createPhysicalOperatorFromUpdate(ctx, op)
+	case *abstract.Delete:
+		return createPhysicalOperatorFromDelete(ctx, op)
+	default:
+		return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "invalid operator tree: %T", op)
+	}
+}
 
-		vindexTable := op.TableInfo.GetVindexTable()
-		opCode := engine.Unsharded
-		if vindexTable.Keyspace.Sharded {
-			opCode = engine.Scatter
-		}
+func createPhysicalOperatorFromUpdate(ctx *plancontext.PlanningContext, op *abstract.Update) (abstract.PhysicalOperator, error) {
+	vindexTable, opCode, dest, err := buildVindexTableForDML(ctx, op.TableInfo, op.Table, "update")
+	if err != nil {
+		return nil, err
+	}
 
-		tblName, ok := op.Table.Alias.Expr.(sqlparser.TableName)
-		if ok {
-			var err error
-			_, _, _, typ, dest, err = ctx.VSchema.FindTableOrVindex(tblName)
-			if err != nil {
-				return nil, err
-			}
-			if dest != nil {
-				if typ != topodatapb.TabletType_PRIMARY {
-					return nil, vterrors.NewErrorf(vtrpcpb.Code_FAILED_PRECONDITION, vterrors.InnodbReadOnly, "unsupported: update statement with a replica target")
-				}
-				// we are dealing with an explicitly targeted UPDATE
-				opCode = engine.ByDestination
-			}
-		}
+	vp, cvv, ovq, err := getUpdateVindexInformation(op, vindexTable)
+	if err != nil {
+		return nil, err
+	}
+
+	r := &Route{
+		Source: &Update{
+			QTable:              op.Table,
+			VTable:              vindexTable,
+			Assignments:         op.Assignments,
+			ChangedVindexValues: cvv,
+			OwnedVindexQuery:    ovq,
+			AST:                 op.AST,
+		},
+		RouteOpCode:       opCode,
+		Keyspace:          vindexTable.Keyspace,
+		VindexPreds:       vp,
+		TargetDestination: dest,
+	}
 
-		vp, cvv, ovq, err := getUpdateVindexInformation(op, vindexTable)
+	for _, predicate := range op.Table.Predicates {
+		err := r.UpdateRoutingLogic(ctx, predicate)
 		if err != nil {
 			return nil, err
 		}
+	}
 
-		r := &Route{
-			Source: &Update{
-				QTable:              op.Table,
-				VTable:              vindexTable,
-				Assignments:         op.Assignments,
-				ChangedVindexValues: cvv,
-				OwnedVindexQuery:    ovq,
-				AST:                 op.AST,
-			},
-			RouteOpCode:       opCode,
-			Keyspace:          vindexTable.Keyspace,
-			VindexPreds:       vp,
-			TargetDestination: dest,
-		}
+	if r.RouteOpCode == engine.Scatter && op.AST.Limit != nil {
+		// TODO systay: we should probably check for other op code types - IN could also hit multiple shards (2022-04-07)
+		return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "multi shard update with limit is not supported")
+	}
 
-		for _, predicate := range op.Table.Predicates {
-			err := r.UpdateRoutingLogic(ctx, predicate)
-			if err != nil {
-				return nil, err
+	return r, nil
+}
+
+func buildVindexTableForDML(ctx *plancontext.PlanningContext, tableInfo semantics.TableInfo, table *abstract.QueryTable, dmlType string) (*vindexes.Table, engine.Opcode, key.Destination, error) {
+	vindexTable := tableInfo.GetVindexTable()
+	opCode := engine.Unsharded
+	if vindexTable.Keyspace.Sharded {
+		opCode = engine.Scatter
+	}
+
+	var dest key.Destination
+	var typ topodatapb.TabletType
+	var err error
+	tblName, ok := table.Alias.Expr.(sqlparser.TableName)
+	if ok {
+		_, _, _, typ, dest, err = ctx.VSchema.FindTableOrVindex(tblName)
+		if err != nil {
+			return nil, 0, nil, err
+		}
+		if dest != nil {
+			if typ != topodatapb.TabletType_PRIMARY {
+				return nil, 0, nil, vterrors.NewErrorf(vtrpcpb.Code_FAILED_PRECONDITION, vterrors.InnodbReadOnly, "unsupported: %v statement with a replica target", dmlType)
 			}
+			// we are dealing with an explicitly targeted UPDATE
+			opCode = engine.ByDestination
 		}
+	}
+	return vindexTable, opCode, dest, nil
+}
+
+func createPhysicalOperatorFromDelete(ctx *plancontext.PlanningContext, op *abstract.Delete) (*Route, error) {
+	var ovq string
+	vindexTable, opCode, dest, err := buildVindexTableForDML(ctx, op.TableInfo, op.Table, "delete")
+	if err != nil {
+		return nil, err
+	}
+
+	primaryVindex, vindexAndPredicates, err := getVindexInformation(op.TableID(), op.Table.Predicates, vindexTable)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(vindexTable.Owned) > 0 {
+		tblExpr := &sqlparser.AliasedTableExpr{Expr: sqlparser.TableName{Name: vindexTable.Name}, As: op.Table.Alias.As}
+		ovq = generateOwnedVindexQuery(tblExpr, op.AST, vindexTable, primaryVindex.Columns)
+	}
 
-		if r.RouteOpCode == engine.Scatter && op.AST.Limit != nil {
-			// TODO systay: we should probably check for other op code types - IN could also hit multiple shards (2022-04-07)
-			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "multi shard update with limit is not supported")
+	r := &Route{
+		Source: &Delete{
+			QTable:           op.Table,
+			VTable:           vindexTable,
+			OwnedVindexQuery: ovq,
+			AST:              op.AST,
+		},
+		RouteOpCode:       opCode,
+		Keyspace:          vindexTable.Keyspace,
+		VindexPreds:       vindexAndPredicates,
+		TargetDestination: dest,
+	}
+
+	for _, predicate := range op.Table.Predicates {
+		err := r.UpdateRoutingLogic(ctx, predicate)
+		if err != nil {
+			return nil, err
 		}
+	}
 
-		return r, nil
-	default:
-		return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "invalid operator tree: %T", op)
+	if r.RouteOpCode == engine.Scatter && op.AST.Limit != nil {
+		// TODO systay: we should probably check for other op code types - IN could also hit multiple shards (2022-04-07)
+		return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "multi shard delete with limit is not supported")
+	}
+	return r, nil
+}
+
+func generateOwnedVindexQuery(tblExpr sqlparser.TableExpr, del *sqlparser.Delete, table *vindexes.Table, ksidCols []sqlparser.IdentifierCI) string {
+	buf := sqlparser.NewTrackedBuffer(nil)
+	for idx, col := range ksidCols {
+		if idx == 0 {
+			buf.Myprintf("select %v", col)
+		} else {
+			buf.Myprintf(", %v", col)
+		}
+	}
+	for _, cv := range table.Owned {
+		for _, column := range cv.Columns {
+			buf.Myprintf(", %v", column)
+		}
 	}
+	buf.Myprintf(" from %v%v%v%v for update", tblExpr, del.Where, del.OrderBy, del.Limit)
+	return buf.String()
 }
 
 func getUpdateVindexInformation(op *abstract.Update, vindexTable *vindexes.Table) ([]*VindexPlusPredicates, map[string]*engine.VindexValues, string, error) {
diff --git a/go/vt/vtgate/semantics/scoper.go b/go/vt/vtgate/semantics/scoper.go
index 97ec5f38b7..5d5f1c8510 100644
--- a/go/vt/vtgate/semantics/scoper.go
+++ b/go/vt/vtgate/semantics/scoper.go
@@ -60,11 +60,11 @@ func newScoper() *scoper {
 func (s *scoper) down(cursor *sqlparser.Cursor) error {
 	node := cursor.Node()
 	switch node := node.(type) {
-	case *sqlparser.Update:
+	case *sqlparser.Update, *sqlparser.Delete:
 		currScope := newScope(s.currentScope())
 		s.push(currScope)
 
-		currScope.stmt = node
+		currScope.stmt = node.(sqlparser.Statement)
 	case *sqlparser.Select:
 		currScope := newScope(s.currentScope())
 		s.push(currScope)
diff --git a/go/vt/vttablet/tabletserver/query_engine.go b/go/vt/vttablet/tabletserver/query_engine.go
index 132526512b..b3e24c7017 100644
--- a/go/vt/vttablet/tabletserver/query_engine.go
+++ b/go/vt/vttablet/tabletserver/query_engine.go
@@ -27,6 +27,9 @@ import (
 	"sync/atomic"
 	"time"
 
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/vterrors"
+
 	"vitess.io/vitess/go/acl"
 	"vitess.io/vitess/go/cache"
 	"vitess.io/vitess/go/mysql"
@@ -96,21 +99,25 @@ func (ep *TabletPlan) buildAuthorized() {
 	}
 }
 
-func (ep *TabletPlan) IsValid(hasReservedCon, hasSysSettings bool) bool {
+func (ep *TabletPlan) IsValid(hasReservedCon, hasSysSettings bool) error {
 	if !ep.NeedsReservedConn {
-		return true
+		return nil
 	}
-	switch ep.PlanID {
+	return isValid(ep.PlanID, hasReservedCon, hasSysSettings)
+}
+
+func isValid(planType planbuilder.PlanType, hasReservedCon bool, hasSysSettings bool) error {
+	switch planType {
 	case planbuilder.PlanSelectLockFunc, planbuilder.PlanDDL:
 		if hasReservedCon {
-			return true
+			return nil
 		}
 	case planbuilder.PlanSet:
 		if hasReservedCon || hasSysSettings {
-			return true
+			return nil
 		}
 	}
-	return false
+	return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "%s not allowed without reserved connection", planType.String())
 }
 
 // _______________________________________________
diff --git a/go/vt/vttablet/tabletserver/tabletenv/config.go b/go/vt/vttablet/tabletserver/tabletenv/config.go
index 19212d710d..e8bb869392 100644
--- a/go/vt/vttablet/tabletserver/tabletenv/config.go
+++ b/go/vt/vttablet/tabletserver/tabletenv/config.go
@@ -173,7 +173,7 @@ func registerTabletEnvFlags(fs *pflag.FlagSet) {
 	fs.BoolVar(&enableReplicationReporter, "enable_replication_reporter", false, "Use polling to track replication lag.")
 	fs.BoolVar(&currentConfig.EnableOnlineDDL, "queryserver_enable_online_ddl", true, "Enable online DDL.")
 	fs.BoolVar(&currentConfig.SanitizeLogMessages, "sanitize_log_messages", false, "Remove potentially sensitive information in tablet INFO, WARNING, and ERROR log messages such as query parameters.")
-	fs.BoolVar(&currentConfig.EnableSettingsPool, "queryserver_enable_settings_pool", false, "Enable pooling of connections with modified system settings")
+	fs.BoolVar(&currentConfig.EnableSettingsPool, "queryserver-enable-settings-pool", false, "Enable pooling of connections with modified system settings")
 
 	fs.Int64Var(&currentConfig.RowStreamer.MaxInnoDBTrxHistLen, "vreplication_copy_phase_max_innodb_history_list_length", 1000000, "The maximum InnoDB transaction history that can exist on a vstreamer (source) before starting another round of copying rows. This helps to limit the impact on the source tablet.")
 	fs.Int64Var(&currentConfig.RowStreamer.MaxMySQLReplLagSecs, "vreplication_copy_phase_max_mysql_replication_lag", 43200, "The maximum MySQL replication lag (in seconds) that can exist on a vstreamer (source) before starting another round of copying rows. This helps to limit the impact on the source tablet.")
diff --git a/go/vt/vttablet/tabletserver/tabletserver.go b/go/vt/vttablet/tabletserver/tabletserver.go
index 25f465e5e5..c00054157c 100644
--- a/go/vt/vttablet/tabletserver/tabletserver.go
+++ b/go/vt/vttablet/tabletserver/tabletserver.go
@@ -750,8 +750,8 @@ func (tsv *TabletServer) execute(ctx context.Context, target *querypb.Target, sq
 			if err != nil {
 				return err
 			}
-			if !plan.IsValid(reservedID != 0, len(settings) > 0) {
-				return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "%s not allowed without reserved connection", plan.PlanID.String())
+			if err = plan.IsValid(reservedID != 0, len(settings) > 0); err != nil {
+				return err
 			}
 			// If both the values are non-zero then by design they are same value. So, it is safe to overwrite.
 			connID := reservedID
@@ -855,8 +855,8 @@ func (tsv *TabletServer) streamExecute(ctx context.Context, target *querypb.Targ
 			if err != nil {
 				return err
 			}
-			if !plan.IsValid(reservedID != 0, len(settings) > 0) {
-				return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "%s not allowed without reserved connection", plan.PlanID.String())
+			if err = plan.IsValid(reservedID != 0, len(settings) > 0); err != nil {
+				return err
 			}
 			// If both the values are non-zero then by design they are same value. So, it is safe to overwrite.
 			connID := reservedID


Test patch:
diff --git a/go/test/endtoend/vreplication/helper_test.go b/go/test/endtoend/vreplication/helper_test.go
index 2c14e07070..264c748a5b 100644
--- a/go/test/endtoend/vreplication/helper_test.go
+++ b/go/test/endtoend/vreplication/helper_test.go
@@ -47,7 +47,11 @@ import (
 const (
 	defaultTick          = 1 * time.Second
 	defaultTimeout       = 30 * time.Second
-	workflowStartTimeout = 5 * time.Second
+	workflowStateTimeout = 90 * time.Second
+	workflowStateCopying = "Copying" // nolint
+	workflowStateRunning = "Running" // nolint
+	workflowStateStopped = "Stopped" // nolint
+	workflowStateError   = "Error"   // nolint
 )
 
 func execMultipleQueries(t *testing.T, conn *mysql.Conn, database string, lines string) {
@@ -234,10 +238,10 @@ func validateThatQueryExecutesOnTablet(t *testing.T, conn *mysql.Conn, tablet *c
 	return newCount == count+1
 }
 
-func waitForWorkflowToStart(t *testing.T, vc *VitessCluster, ksWorkflow string) {
+func waitForWorkflowState(t *testing.T, vc *VitessCluster, ksWorkflow string, wantState string) {
 	done := false
-	timer := time.NewTimer(workflowStartTimeout)
-	log.Infof("Waiting for workflow %s to start", ksWorkflow)
+	timer := time.NewTimer(workflowStateTimeout)
+	log.Infof("Waiting for workflow %q to fully reach %q state", ksWorkflow, wantState)
 	for {
 		output, err := vc.VtctlClient.ExecuteCommandWithOutput("Workflow", ksWorkflow, "show")
 		require.NoError(t, err)
@@ -249,8 +253,8 @@ func waitForWorkflowToStart(t *testing.T, vc *VitessCluster, ksWorkflow string)
 				if streamId.String() == "PrimaryReplicationStatuses" {
 					streamInfos.ForEach(func(attributeKey, attributeValue gjson.Result) bool { // for each attribute in the stream
 						state = attributeValue.Get("State").String()
-						if state != "Running" {
-							done = false // we need to wait for all streams to start
+						if state != wantState {
+							done = false // we need to wait for all streams to have the desired state
 						}
 						return true
 					})
@@ -260,13 +264,13 @@ func waitForWorkflowToStart(t *testing.T, vc *VitessCluster, ksWorkflow string)
 			return true
 		})
 		if done {
-			log.Infof("Workflow %s has started", ksWorkflow)
+			log.Infof("Workflow %q has fully reached the desired state of %q", ksWorkflow, wantState)
 			return
 		}
 		select {
 		case <-timer.C:
-			require.FailNowf(t, "workflow %q did not fully start before the timeout of %s",
-				ksWorkflow, workflowStartTimeout)
+			require.FailNowf(t, "workflow %q did not fully reach the expected state of %q before the timeout of %s; last seen output: %s",
+				ksWorkflow, wantState, workflowStateTimeout, output)
 		default:
 			time.Sleep(defaultTick)
 		}
@@ -463,3 +467,39 @@ func getDebugVar(t *testing.T, port int, varPath []string) (string, error) {
 	require.NoError(t, err)
 	return string(val), nil
 }
+
+func confirmWorkflowHasCopiedNoData(t *testing.T, targetKS, workflow string) {
+	timer := time.NewTimer(defaultTimeout)
+	defer timer.Stop()
+	ksWorkflow := fmt.Sprintf("%s.%s", targetKS, workflow)
+	for {
+		output, err := vc.VtctlClient.ExecuteCommandWithOutput("Workflow", ksWorkflow, "show")
+		require.NoError(t, err)
+		result := gjson.Get(output, "ShardStatuses")
+		result.ForEach(func(tabletId, tabletStreams gjson.Result) bool { // for each source tablet
+			tabletStreams.ForEach(func(streamId, streamInfos gjson.Result) bool { // for each stream
+				if streamId.String() == "PrimaryReplicationStatuses" {
+					streamInfos.ForEach(func(attributeKey, attributeValue gjson.Result) bool { // for each attribute in the stream
+						state := attributeValue.Get("State").String()
+						pos := attributeValue.Get("Pos").String()
+						// If we've actually copied anything then we'll have a position in the stream
+						if (state == workflowStateRunning || state == workflowStateCopying) && pos != "" {
+							require.FailNowf(t, "Unexpected data copied in workflow",
+								"The MoveTables workflow %q copied data in less than %s when it should have been waiting. Show output: %s",
+								ksWorkflow, defaultTimeout, output)
+						}
+						return true // end attribute loop
+					})
+				}
+				return true // end stream loop
+			})
+			return true // end tablet loop
+		})
+		select {
+		case <-timer.C:
+			return
+		default:
+			time.Sleep(defaultTick)
+		}
+	}
+}
diff --git a/go/test/endtoend/vreplication/migrate_test.go b/go/test/endtoend/vreplication/migrate_test.go
index cfedfce2f6..18745aea4c 100644
--- a/go/test/endtoend/vreplication/migrate_test.go
+++ b/go/test/endtoend/vreplication/migrate_test.go
@@ -107,7 +107,7 @@ func TestMigrate(t *testing.T) {
 			"--source=ext1.rating", "create", ksWorkflow); err != nil {
 			t.Fatalf("Migrate command failed with %+v : %s\n", err, output)
 		}
-		waitForWorkflowToStart(t, vc, ksWorkflow)
+		waitForWorkflowState(t, vc, ksWorkflow, workflowStateRunning)
 		expectNumberOfStreams(t, vtgateConn, "migrate", "e1", "product:0", 1)
 		waitForRowCount(t, vtgateConn, "product:0", "rating", 2)
 		waitForRowCount(t, vtgateConn, "product:0", "review", 3)
diff --git a/go/test/endtoend/vreplication/resharding_workflows_v2_test.go b/go/test/endtoend/vreplication/resharding_workflows_v2_test.go
index a5680eab42..74c1739a06 100644
--- a/go/test/endtoend/vreplication/resharding_workflows_v2_test.go
+++ b/go/test/endtoend/vreplication/resharding_workflows_v2_test.go
@@ -60,7 +60,7 @@ func createReshardWorkflow(t *testing.T, sourceShards, targetShards string) erro
 	err := tstWorkflowExec(t, defaultCellName, workflowName, targetKs, targetKs,
 		"", workflowActionCreate, "", sourceShards, targetShards)
 	require.NoError(t, err)
-	waitForWorkflowToStart(t, vc, ksWorkflow)
+	waitForWorkflowState(t, vc, ksWorkflow, workflowStateRunning)
 	catchup(t, targetTab1, workflowName, "Reshard")
 	catchup(t, targetTab2, workflowName, "Reshard")
 	vdiff1(t, ksWorkflow, "")
@@ -74,7 +74,7 @@ func createMoveTablesWorkflow(t *testing.T, tables string) error {
 	err := tstWorkflowExec(t, defaultCellName, workflowName, sourceKs, targetKs,
 		tables, workflowActionCreate, "", "", "")
 	require.NoError(t, err)
-	waitForWorkflowToStart(t, vc, ksWorkflow)
+	waitForWorkflowState(t, vc, ksWorkflow, workflowStateRunning)
 	catchup(t, targetTab1, workflowName, "MoveTables")
 	catchup(t, targetTab2, workflowName, "MoveTables")
 	vdiff1(t, ksWorkflow, "")
@@ -267,7 +267,7 @@ func testVSchemaForSequenceAfterMoveTables(t *testing.T) {
 		"customer2", workflowActionCreate, "", "", "")
 	require.NoError(t, err)
 
-	waitForWorkflowToStart(t, vc, "customer.wf2")
+	waitForWorkflowState(t, vc, "customer.wf2", workflowStateRunning)
 	waitForLowLag(t, "customer", "wf2")
 
 	err = tstWorkflowExec(t, defaultCellName, "wf2", sourceKs, targetKs,
@@ -301,7 +301,7 @@ func testVSchemaForSequenceAfterMoveTables(t *testing.T) {
 	err = tstWorkflowExec(t, defaultCellName, "wf3", targetKs, sourceKs,
 		"customer2", workflowActionCreate, "", "", "")
 	require.NoError(t, err)
-	waitForWorkflowToStart(t, vc, "product.wf3")
+	waitForWorkflowState(t, vc, "product.wf3", workflowStateRunning)
 
 	waitForLowLag(t, "product", "wf3")
 	err = tstWorkflowExec(t, defaultCellName, "wf3", targetKs, sourceKs,
diff --git a/go/test/endtoend/vreplication/vreplication_test.go b/go/test/endtoend/vreplication/vreplication_test.go
index f4cce28db7..b2e456d279 100644
--- a/go/test/endtoend/vreplication/vreplication_test.go
+++ b/go/test/endtoend/vreplication/vreplication_test.go
@@ -36,7 +36,6 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/buger/jsonparser"
-	"github.com/tidwall/gjson"
 
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/sqltypes"
@@ -149,7 +148,10 @@ func TestVreplicationCopyThrottling(t *testing.T) {
 	waitForInnoDBHistoryLength(t, vc.getPrimaryTablet(t, sourceKs, shard), maxSourceTrxHistory)
 	// We need to force primary tablet types as the history list has been increased on the source primary
 	moveTablesWithTabletTypes(t, defaultCell.Name, workflow, sourceKs, targetKs, table, "primary")
-	verifySourceTabletThrottling(t, targetKs, workflow)
+	// Wait for the copy phase to start
+	waitForWorkflowState(t, vc, fmt.Sprintf("%s.%s", targetKs, workflow), workflowStateCopying)
+	// The initial copy phase should be blocking on the history list
+	confirmWorkflowHasCopiedNoData(t, targetKs, workflow)
 	releaseInnoDBRowHistory(t, trxConn)
 	trxConn.Close()
 }
@@ -653,40 +655,6 @@ func validateRollupReplicates(t *testing.T) {
 	})
 }
 
-func verifySourceTabletThrottling(t *testing.T, targetKS, workflow string) {
-	timer := time.NewTimer(defaultTimeout)
-	defer timer.Stop()
-	ksWorkflow := fmt.Sprintf("%s.%s", targetKS, workflow)
-	for {
-		output, err := vc.VtctlClient.ExecuteCommandWithOutput("Workflow", ksWorkflow, "show")
-		require.NoError(t, err)
-		result := gjson.Get(output, "ShardStatuses")
-		result.ForEach(func(tabletId, tabletStreams gjson.Result) bool { // for each source tablet
-			tabletStreams.ForEach(func(streamId, streamInfos gjson.Result) bool { // for each stream
-				if streamId.String() == "PrimaryReplicationStatuses" {
-					streamInfos.ForEach(func(attributeKey, attributeValue gjson.Result) bool { // for each attribute in the stream
-						state := attributeValue.Get("State").String()
-						if state != "Copying" {
-							require.FailNowf(t, "Unexpected running workflow stream",
-								"Initial copy phase for the MoveTables workflow %s started in less than %s when it should have been waiting. Show output: %s",
-								ksWorkflow, defaultTimeout, output)
-						}
-						return true // end attribute loop
-					})
-				}
-				return true // end stream loop
-			})
-			return true // end tablet loop
-		})
-		select {
-		case <-timer.C:
-			return
-		default:
-			time.Sleep(defaultTick)
-		}
-	}
-}
-
 func reshardCustomer2to4Split(t *testing.T, cells []*Cell, sourceCellOrAlias string) {
 	t.Run("reshardCustomer2to4Split", func(t *testing.T) {
 		ksName := "customer"
@@ -1217,8 +1185,9 @@ func generateInnoDBRowHistory(t *testing.T, sourceKS string, neededTrxHistory in
 	execQuery(t, dbConn1, "use "+sourceKS)
 	execQuery(t, dbConn2, "use "+sourceKS)
 	offset := int64(1000)
+	limit := int64(neededTrxHistory * 100)
 	insertStmt := strings.Builder{}
-	for i := offset; i <= (neededTrxHistory*10)+offset; i++ {
+	for i := offset; i <= offset+limit; i++ {
 		if i == offset {
 			insertStmt.WriteString(fmt.Sprintf("insert into product (pid, description) values (%d, 'test')", i))
 		} else {
@@ -1231,10 +1200,12 @@ func generateInnoDBRowHistory(t *testing.T, sourceKS string, neededTrxHistory in
 	execQuery(t, dbConn2, "rollback")
 	execQuery(t, dbConn2, "start transaction")
 	execQuery(t, dbConn2, "select count(*) from product")
-	execQuery(t, dbConn1, fmt.Sprintf("delete from product where pid >= %d and pid < %d", offset, offset+10000))
+	execQuery(t, dbConn1, fmt.Sprintf("delete from product where pid >= %d and pid <= %d", offset, offset+limit))
 	return dbConn2
 }
 
+// waitForInnoDBRowHistory waits for the history list length to be greater than the
+// expected length.
 func waitForInnoDBHistoryLength(t *testing.T, tablet *cluster.VttabletProcess, expectedLength int64) {
 	timer := time.NewTimer(defaultTimeout)
 	historyLen := int64(0)
@@ -1245,7 +1216,7 @@ func waitForInnoDBHistoryLength(t *testing.T, tablet *cluster.VttabletProcess, e
 		require.Equal(t, 1, len(res.Rows))
 		historyLen, err = res.Rows[0][0].ToInt64()
 		require.NoError(t, err)
-		if historyLen >= expectedLength {
+		if historyLen > expectedLength {
 			return
 		}
 		select {
diff --git a/go/test/endtoend/vtgate/reservedconn/main_test.go b/go/test/endtoend/vtgate/reservedconn/main_test.go
index 4784689d90..73044fe457 100644
--- a/go/test/endtoend/vtgate/reservedconn/main_test.go
+++ b/go/test/endtoend/vtgate/reservedconn/main_test.go
@@ -134,7 +134,7 @@ func runAllTests(m *testing.M) int {
 	}
 	clusterInstance.VtTabletExtraArgs = []string{"--queryserver-config-transaction-timeout", "5"}
 	if enableSettingsPool {
-		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver_enable_settings_pool")
+		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver-enable-settings-pool")
 	}
 	if err := clusterInstance.StartKeyspace(*keyspace, []string{"-80", "80-"}, 1, false); err != nil {
 		return 1
diff --git a/go/test/endtoend/vtgate/reservedconn/reconnect1/main_test.go b/go/test/endtoend/vtgate/reservedconn/reconnect1/main_test.go
index 978ffec669..577155f16c 100644
--- a/go/test/endtoend/vtgate/reservedconn/reconnect1/main_test.go
+++ b/go/test/endtoend/vtgate/reservedconn/reconnect1/main_test.go
@@ -97,7 +97,7 @@ func runAllTests(m *testing.M) int {
 		VSchema:   vSchema,
 	}
 	if enableSettingsPool {
-		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver_enable_settings_pool")
+		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver-enable-settings-pool")
 	}
 	if err := clusterInstance.StartKeyspace(*keyspace, []string{"-80", "80-"}, 1, true); err != nil {
 		return 1
diff --git a/go/test/endtoend/vtgate/reservedconn/reconnect2/main_test.go b/go/test/endtoend/vtgate/reservedconn/reconnect2/main_test.go
index fa3002e5a6..f97d96ef89 100644
--- a/go/test/endtoend/vtgate/reservedconn/reconnect2/main_test.go
+++ b/go/test/endtoend/vtgate/reservedconn/reconnect2/main_test.go
@@ -98,7 +98,7 @@ func runAllTests(m *testing.M) int {
 	}
 	clusterInstance.VtTabletExtraArgs = []string{"--queryserver-config-transaction-timeout", "5"}
 	if enableSettingsPool {
-		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver_enable_settings_pool")
+		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver-enable-settings-pool")
 	}
 	if err := clusterInstance.StartKeyspace(*keyspace, []string{"-80", "80-"}, 1, true); err != nil {
 		return 1
diff --git a/go/test/endtoend/vtgate/reservedconn/reconnect3/main_test.go b/go/test/endtoend/vtgate/reservedconn/reconnect3/main_test.go
index 5e88ccf91b..25af85acc0 100644
--- a/go/test/endtoend/vtgate/reservedconn/reconnect3/main_test.go
+++ b/go/test/endtoend/vtgate/reservedconn/reconnect3/main_test.go
@@ -72,7 +72,7 @@ func runAllTests(m *testing.M) int {
 		SchemaSQL: sqlSchema,
 	}
 	if enableSettingsPool {
-		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver_enable_settings_pool")
+		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver-enable-settings-pool")
 	}
 	if err := clusterInstance.StartUnshardedKeyspace(*keyspace, 2, false); err != nil {
 		return 1
diff --git a/go/test/endtoend/vtgate/reservedconn/reconnect4/main_test.go b/go/test/endtoend/vtgate/reservedconn/reconnect4/main_test.go
index 02733eaac5..28367cd597 100644
--- a/go/test/endtoend/vtgate/reservedconn/reconnect4/main_test.go
+++ b/go/test/endtoend/vtgate/reservedconn/reconnect4/main_test.go
@@ -72,7 +72,7 @@ func runAllTests(m *testing.M) int {
 		SchemaSQL: sqlSchema,
 	}
 	if enableSettingsPool {
-		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver_enable_settings_pool")
+		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver-enable-settings-pool")
 	}
 	if err := clusterInstance.StartUnshardedKeyspace(*keyspace, 2, false); err != nil {
 		return 1
diff --git a/go/test/endtoend/vtgate/unsharded/main_test.go b/go/test/endtoend/vtgate/unsharded/main_test.go
index f7b3fb43ab..1f25db161e 100644
--- a/go/test/endtoend/vtgate/unsharded/main_test.go
+++ b/go/test/endtoend/vtgate/unsharded/main_test.go
@@ -182,7 +182,7 @@ func runAllTests(m *testing.M) int {
 	}
 	clusterInstance.VtTabletExtraArgs = []string{"--queryserver-config-transaction-timeout", "3", "--queryserver-config-max-result-size", "30"}
 	if enableSettingsPool {
-		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver_enable_settings_pool")
+		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver-enable-settings-pool")
 	}
 	if err := clusterInstance.StartUnshardedKeyspace(*Keyspace, 0, false); err != nil {
 		log.Fatal(err.Error())
diff --git a/go/vt/vtgate/planbuilder/testdata/dml_cases.txt b/go/vt/vtgate/planbuilder/testdata/dml_cases.txt
index 2d0f394e04..1fa43a7c95 100644
--- a/go/vt/vtgate/planbuilder/testdata/dml_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/dml_cases.txt
@@ -833,7 +833,32 @@ Gen4 plan same as above
     "user.user"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from user where id = 1",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "KsidLength": 1,
+    "KsidVindex": "user_index",
+    "MultiShardAutocommit": false,
+    "OwnedVindexQuery": "select Id, `Name`, Costly from `user` where id = 1 for update",
+    "Query": "delete from `user` where id = 1",
+    "Table": "user",
+    "Values": [
+      "INT64(1)"
+    ],
+    "Vindex": "user_index"
+  },
+  "TablesUsed": [
+    "user.user"
+  ]
+}
 
 # multi-table delete with comma join
 "delete a from unsharded_a a, unsharded_b b where a.id = b.id and b.val = 1"
@@ -934,7 +959,32 @@ Gen4 plan same as above
     "user.user"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from route1 where id = 1",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "KsidLength": 1,
+    "KsidVindex": "user_index",
+    "MultiShardAutocommit": false,
+    "OwnedVindexQuery": "select Id, `Name`, Costly from `user` as route1 where id = 1 for update",
+    "Query": "delete from `user` as route1 where id = 1",
+    "Table": "user",
+    "Values": [
+      "INT64(1)"
+    ],
+    "Vindex": "user_index"
+  },
+  "TablesUsed": [
+    "user.user"
+  ]
+}
 
 # delete: routing rules for subquery
 "delete from  unsharded_a where a=(select a from route2)"
@@ -957,7 +1007,26 @@ Gen4 plan same as above
     "main.unsharded_a"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from  unsharded_a where a=(select a from route2)",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "Unsharded",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "TargetTabletType": "PRIMARY",
+    "MultiShardAutocommit": false,
+    "Query": "delete from unsharded_a where a = (select a from unsharded as route2)",
+    "Table": "unsharded, unsharded_a"
+  },
+  "TablesUsed": [
+    "main.unsharded",
+    "main.unsharded_a"
+  ]
+}
 
 # update by lookup
 "update music set val = 1 where id = 1"
@@ -1084,7 +1153,32 @@ Gen4 plan same as above
     "user.music"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from music where id = 1",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "KsidLength": 1,
+    "KsidVindex": "user_index",
+    "MultiShardAutocommit": false,
+    "OwnedVindexQuery": "select user_id, id from music where id = 1 for update",
+    "Query": "delete from music where id = 1",
+    "Table": "music",
+    "Values": [
+      "INT64(1)"
+    ],
+    "Vindex": "music_user_map"
+  },
+  "TablesUsed": [
+    "user.music"
+  ]
+}
 
 # delete from, no owned vindexes
 "delete from music_extra where user_id = 1"
@@ -1111,7 +1205,29 @@ Gen4 plan same as above
     "user.music_extra"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from music_extra where user_id = 1",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "MultiShardAutocommit": false,
+    "Query": "delete from music_extra where user_id = 1",
+    "Table": "music_extra",
+    "Values": [
+      "INT64(1)"
+    ],
+    "Vindex": "user_index"
+  },
+  "TablesUsed": [
+    "user.music_extra"
+  ]
+}
 
 # simple insert, no values
 "insert into unsharded values()"
@@ -2344,7 +2460,32 @@ Gen4 plan same as above
     "user.multicolvin"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from multicolvin where kid=1",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "KsidLength": 1,
+    "KsidVindex": "kid_index",
+    "MultiShardAutocommit": false,
+    "OwnedVindexQuery": "select kid, column_a, column_b, column_c from multicolvin where kid = 1 for update",
+    "Query": "delete from multicolvin where kid = 1",
+    "Table": "multicolvin",
+    "Values": [
+      "INT64(1)"
+    ],
+    "Vindex": "kid_index"
+  },
+  "TablesUsed": [
+    "user.multicolvin"
+  ]
+}
 
 # update columns of multi column vindex
 "update multicolvin set column_b = 1, column_c = 2 where kid = 1"
@@ -2912,7 +3053,26 @@ Gen4 plan same as above
     "main.unsharded"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from unsharded where col = (select id from unsharded_a where id = unsharded.col)",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "Unsharded",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "TargetTabletType": "PRIMARY",
+    "MultiShardAutocommit": false,
+    "Query": "delete from unsharded where col = (select id from unsharded_a where id = unsharded.col)",
+    "Table": "unsharded, unsharded_a"
+  },
+  "TablesUsed": [
+    "main.unsharded",
+    "main.unsharded_a"
+  ]
+}
 
 # update vindex value to null
 "update user set name = null where id = 1"
@@ -3199,7 +3359,32 @@ Gen4 plan same as above
     "user.music"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete music from music where id = 1",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "KsidLength": 1,
+    "KsidVindex": "user_index",
+    "MultiShardAutocommit": false,
+    "OwnedVindexQuery": "select user_id, id from music where id = 1 for update",
+    "Query": "delete from music where id = 1",
+    "Table": "music",
+    "Values": [
+      "INT64(1)"
+    ],
+    "Vindex": "music_user_map"
+  },
+  "TablesUsed": [
+    "user.music"
+  ]
+}
 
 # scatter update table with owned vindexes without changing lookup vindex
 "update user set val = 1"
@@ -3396,7 +3581,32 @@ Gen4 plan same as above
     "user.user"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from user where name = _binary 'abc'",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "Equal",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "KsidLength": 1,
+    "KsidVindex": "user_index",
+    "MultiShardAutocommit": false,
+    "OwnedVindexQuery": "select Id, `Name`, Costly from `user` where `name` = _binary 'abc' for update",
+    "Query": "delete from `user` where `name` = _binary 'abc'",
+    "Table": "user",
+    "Values": [
+      "VARBINARY(\"abc\")"
+    ],
+    "Vindex": "name_user_map"
+  },
+  "TablesUsed": [
+    "user.user"
+  ]
+}
 
 # delete with shard targeting
 "delete from `user[-]`.user"
@@ -3622,7 +3832,32 @@ Gen4 plan same as above
     "zlookup_unique.t1"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from zlookup_unique.t1 where c2 = 10 and c3 = 20",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "zlookup_unique",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "KsidLength": 1,
+    "KsidVindex": "xxhash",
+    "MultiShardAutocommit": false,
+    "OwnedVindexQuery": "select c1, c2, c3 from t1 where c2 = 10 and c3 = 20 for update",
+    "Query": "delete from t1 where c2 = 10 and c3 = 20",
+    "Table": "t1",
+    "Values": [
+      "INT64(20)"
+    ],
+    "Vindex": "lookup_t1_2"
+  },
+  "TablesUsed": [
+    "zlookup_unique.t1"
+  ]
+}
 
 # Update on backfilling and non-backfilling unique lookup vindexes should be an equal
 "update zlookup_unique.t1 set c2 = 1 where c2 = 10 and c3 = 20"
@@ -3990,7 +4225,33 @@ Gen4 plan same as above
     "user.multicol_tbl"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from multicol_tbl where cola = 1 and colb = 2",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "KsidLength": 2,
+    "KsidVindex": "multicolIdx",
+    "MultiShardAutocommit": false,
+    "OwnedVindexQuery": "select cola, colb, colc, `name` from multicol_tbl where cola = 1 and colb = 2 for update",
+    "Query": "delete from multicol_tbl where cola = 1 and colb = 2",
+    "Table": "multicol_tbl",
+    "Values": [
+      "INT64(1)",
+      "INT64(2)"
+    ],
+    "Vindex": "multicolIdx"
+  },
+  "TablesUsed": [
+    "user.multicol_tbl"
+  ]
+}
 
 # delete with a multicol vindex - reverse order
 "delete from multicol_tbl where colb = 2 and cola = 1"
@@ -4021,7 +4282,33 @@ Gen4 plan same as above
     "user.multicol_tbl"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from multicol_tbl where colb = 2 and cola = 1",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "KsidLength": 2,
+    "KsidVindex": "multicolIdx",
+    "MultiShardAutocommit": false,
+    "OwnedVindexQuery": "select cola, colb, colc, `name` from multicol_tbl where colb = 2 and cola = 1 for update",
+    "Query": "delete from multicol_tbl where colb = 2 and cola = 1",
+    "Table": "multicol_tbl",
+    "Values": [
+      "INT64(1)",
+      "INT64(2)"
+    ],
+    "Vindex": "multicolIdx"
+  },
+  "TablesUsed": [
+    "user.multicol_tbl"
+  ]
+}
 
 # delete with a multicol vindex using an IN clause
 "delete from multicol_tbl where colb IN (1,2) and cola = 1"
@@ -4425,7 +4712,32 @@ Gen4 plan same as above
     "user.multicol_tbl"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from multicol_tbl where cola = 1",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "SubShard",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "KsidLength": 2,
+    "KsidVindex": "multicolIdx",
+    "MultiShardAutocommit": false,
+    "OwnedVindexQuery": "select cola, colb, colc, `name` from multicol_tbl where cola = 1 for update",
+    "Query": "delete from multicol_tbl where cola = 1",
+    "Table": "multicol_tbl",
+    "Values": [
+      "INT64(1)"
+    ],
+    "Vindex": "multicolIdx"
+  },
+  "TablesUsed": [
+    "user.multicol_tbl"
+  ]
+}
 
 # delete with routing using subsharding column with in query
 "delete from multicol_tbl where cola in (1,2)"
@@ -4485,7 +4797,32 @@ Gen4 plan same as above
     "user.multicol_tbl"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "delete from multicol_tbl where name = 'foo' and cola = 2",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "Equal",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "KsidLength": 2,
+    "KsidVindex": "multicolIdx",
+    "MultiShardAutocommit": false,
+    "OwnedVindexQuery": "select cola, colb, colc, `name` from multicol_tbl where `name` = 'foo' and cola = 2 for update",
+    "Query": "delete from multicol_tbl where `name` = 'foo' and cola = 2",
+    "Table": "multicol_tbl",
+    "Values": [
+      "VARCHAR(\"foo\")"
+    ],
+    "Vindex": "name_muticoltbl_map"
+  },
+  "TablesUsed": [
+    "user.multicol_tbl"
+  ]
+}
 
 # insert using select with simple table.
 "insert into music(id, user_id) select * from user"
@@ -5545,7 +5882,6 @@ Gen4 plan same as above
 }
 Gen4 plan same as above
 
-
 # explain dml without any directive should fail
 "explain format=vtexplain delete from user"
 "explain format = vtexplain will actually run queries. `/*vt+ EXECUTE_DML_QUERIES */` must be set to run DML queries in vtexplain. Example: `explain /*vt+ EXECUTE_DML_QUERIES */ format = vtexplain delete from t1`"
@@ -5613,3 +5949,46 @@ Gen4 plan same as above
 }
 Gen4 plan same as above
 
+# Here V3 populates the TablesUsed incorrectly
+# delete with join from multi table join subquery.
+"delete foo from unsharded as foo join (select id from unsharded a join unsharded_b b on a.user_id = b.user_id) as keepers on foo.id = keepers.id where keepers.id is null and foo.col is not null and foo.col \u003c 1000"
+{
+  "QueryType": "DELETE",
+  "Original": "delete foo from unsharded as foo join (select id from unsharded a join unsharded_b b on a.user_id = b.user_id) as keepers on foo.id = keepers.id where keepers.id is null and foo.col is not null and foo.col \u003c 1000",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "Unsharded",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "TargetTabletType": "PRIMARY",
+    "MultiShardAutocommit": false,
+    "Query": "delete foo from unsharded as foo join (select id from unsharded as a join unsharded_b as b on a.user_id = b.user_id) as keepers on foo.id = keepers.id where keepers.id is null and foo.col is not null and foo.col \u003c 1000",
+    "Table": "unsharded, unsharded, unsharded_b"
+  },
+  "TablesUsed": [
+    "main.unsharded",
+    "main.unsharded, unsharded_b"
+  ]
+}
+{
+  "QueryType": "DELETE",
+  "Original": "delete foo from unsharded as foo join (select id from unsharded a join unsharded_b b on a.user_id = b.user_id) as keepers on foo.id = keepers.id where keepers.id is null and foo.col is not null and foo.col \u003c 1000",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "Unsharded",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "TargetTabletType": "PRIMARY",
+    "MultiShardAutocommit": false,
+    "Query": "delete foo from unsharded as foo join (select id from unsharded as a join unsharded_b as b on a.user_id = b.user_id) as keepers on foo.id = keepers.id where keepers.id is null and foo.col is not null and foo.col \u003c 1000",
+    "Table": "unsharded, unsharded_b"
+  },
+  "TablesUsed": [
+    "main.unsharded",
+    "main.unsharded_b"
+  ]
+}
diff --git a/go/vt/vtgate/planbuilder/testdata/oltp_cases.txt b/go/vt/vtgate/planbuilder/testdata/oltp_cases.txt
index 18c09ea319..42aefb3fd4 100644
--- a/go/vt/vtgate/planbuilder/testdata/oltp_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/oltp_cases.txt
@@ -345,7 +345,29 @@
     "main.sbtest15"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "DELETE FROM sbtest15 WHERE id=7525",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "MultiShardAutocommit": false,
+    "Query": "delete from sbtest15 where id = 7525",
+    "Table": "sbtest15",
+    "Values": [
+      "INT64(7525)"
+    ],
+    "Vindex": "hash"
+  },
+  "TablesUsed": [
+    "main.sbtest15"
+  ]
+}
 
 # OLTP insert
 "INSERT INTO sbtest16 (id, k, c, pad) VALUES (42, 1, 2, 50)"
diff --git a/go/vt/vtgate/planbuilder/testdata/tpcc_cases.txt b/go/vt/vtgate/planbuilder/testdata/tpcc_cases.txt
index 7e7c96e033..ced6e2f542 100644
--- a/go/vt/vtgate/planbuilder/testdata/tpcc_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/tpcc_cases.txt
@@ -1126,7 +1126,29 @@ Gen4 plan same as above
     "main.new_orders1"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "DELETE FROM new_orders1 WHERE no_o_id = 2218 AND no_d_id = 358 AND no_w_id = 98465",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "MultiShardAutocommit": false,
+    "Query": "delete from new_orders1 where no_o_id = 2218 and no_d_id = 358 and no_w_id = 98465",
+    "Table": "new_orders1",
+    "Values": [
+      "INT64(98465)"
+    ],
+    "Vindex": "hash"
+  },
+  "TablesUsed": [
+    "main.new_orders1"
+  ]
+}
 
 # TPC-C select unique orders1
 "SELECT o_c_id FROM orders1 WHERE o_id = 6 AND o_d_id = 1983 AND o_w_id = 894605"
@@ -1639,7 +1661,29 @@ Gen4 plan same as above
     "main.order_line1"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "DELETE FROM order_line1 where ol_w_id=178 AND ol_d_id=1 AND ol_o_id=84",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "MultiShardAutocommit": false,
+    "Query": "delete from order_line1 where ol_w_id = 178 and ol_d_id = 1 and ol_o_id = 84",
+    "Table": "order_line1",
+    "Values": [
+      "INT64(178)"
+    ],
+    "Vindex": "hash"
+  },
+  "TablesUsed": [
+    "main.order_line1"
+  ]
+}
 
 # TPC-C delete orders1
 "DELETE FROM orders1 where o_w_id=1 AND o_d_id=3 and o_id=384"
@@ -1666,7 +1710,29 @@ Gen4 plan same as above
     "main.orders1"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "DELETE FROM orders1 where o_w_id=1 AND o_d_id=3 and o_id=384",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "MultiShardAutocommit": false,
+    "Query": "delete from orders1 where o_w_id = 1 and o_d_id = 3 and o_id = 384",
+    "Table": "orders1",
+    "Values": [
+      "INT64(1)"
+    ],
+    "Vindex": "hash"
+  },
+  "TablesUsed": [
+    "main.orders1"
+  ]
+}
 
 # TPC-C delete history1
 "DELETE FROM history1 where h_w_id=75 AND h_d_id=102 LIMIT 10"
@@ -1693,4 +1759,26 @@ Gen4 plan same as above
     "main.history1"
   ]
 }
-Gen4 plan same as above
+{
+  "QueryType": "DELETE",
+  "Original": "DELETE FROM history1 where h_w_id=75 AND h_d_id=102 LIMIT 10",
+  "Instructions": {
+    "OperatorType": "Delete",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": true
+    },
+    "TargetTabletType": "PRIMARY",
+    "MultiShardAutocommit": false,
+    "Query": "delete from history1 where h_w_id = 75 and h_d_id = 102 limit 10",
+    "Table": "history1",
+    "Values": [
+      "INT64(75)"
+    ],
+    "Vindex": "hash"
+  },
+  "TablesUsed": [
+    "main.history1"
+  ]
+}
diff --git a/go/vt/vttablet/tabletserver/planbuilder/plan_test.go b/go/vt/vttablet/tabletserver/planbuilder/plan_test.go
index 4652b5057c..3660648917 100644
--- a/go/vt/vttablet/tabletserver/planbuilder/plan_test.go
+++ b/go/vt/vttablet/tabletserver/planbuilder/plan_test.go
@@ -102,33 +102,6 @@ func TestPlan(t *testing.T) {
 	}
 }
 
-func TestPlanPoolUnsafe(t *testing.T) {
-	testSchema := loadSchema("schema_test.json")
-	for tcase := range iterateExecFile("pool_unsafe_cases.txt") {
-		t.Run(tcase.input, func(t *testing.T) {
-			var plan *Plan
-			var err error
-			statement, err := sqlparser.Parse(tcase.input)
-			require.NoError(t, err)
-			// Plan building will not fail, but it will mark that reserved connection is needed.
-			plan, err = Build(statement, testSchema, "dbName")
-			require.NoError(t, err)
-			require.True(t, plan.NeedsReservedConn)
-			out := fmt.Sprintf("%s not allowed without reserved connection", plan.PlanID.String())
-			if out != tcase.output {
-				t.Errorf("Line:%v\ngot  = %s\nwant = %s", tcase.lineno, out, tcase.output)
-				if err != nil {
-					out = fmt.Sprintf("\"%s\"", out)
-				} else {
-					bout, _ := json.MarshalIndent(plan, "", "  ")
-					out = string(bout)
-				}
-				fmt.Printf("\"%s\"\n%s\n\n", tcase.input, out)
-			}
-		})
-	}
-}
-
 func TestPlanInReservedConn(t *testing.T) {
 	testSchema := loadSchema("schema_test.json")
 	for tcase := range iterateExecFile("exec_cases.txt") {
diff --git a/go/vt/vttablet/tabletserver/planbuilder/testdata/pool_unsafe_cases.txt b/go/vt/vttablet/tabletserver/planbuilder/testdata/pool_unsafe_cases.txt
deleted file mode 100644
index 1ebc5d1508..0000000000
--- a/go/vt/vttablet/tabletserver/planbuilder/testdata/pool_unsafe_cases.txt
+++ /dev/null
@@ -1,15 +0,0 @@
-# get_lock named locks are unsafe with server-side connection pooling
-"select get_lock('foo', 10) from dual"
-"SelectLockFunc not allowed without reserved connection"
-
-# setting system variables must happen inside reserved connections
-"set sql_safe_updates = false"
-"Set not allowed without reserved connection"
-
-# setting system variables must happen inside reserved connections
-"set @@sql_safe_updates = false"
-"Set not allowed without reserved connection"
-
-# setting system variables must happen inside reserved connections
-"set @udv = false"
-"Set not allowed without reserved connection"
diff --git a/go/vt/vttablet/tabletserver/query_engine_test.go b/go/vt/vttablet/tabletserver/query_engine_test.go
index bee2b4c7c0..c03af9186b 100644
--- a/go/vt/vttablet/tabletserver/query_engine_test.go
+++ b/go/vt/vttablet/tabletserver/query_engine_test.go
@@ -32,6 +32,8 @@ import (
 	"testing"
 	"time"
 
+	"vitess.io/vitess/go/vt/sqlparser"
+
 	"vitess.io/vitess/go/mysql"
 
 	"github.com/stretchr/testify/assert"
@@ -669,3 +671,40 @@ func TestAddQueryStats(t *testing.T) {
 		})
 	}
 }
+
+func TestPlanPoolUnsafe(t *testing.T) {
+	tcases := []struct {
+		name, query, err string
+	}{
+		{
+			"get_lock named locks are unsafe with server-side connection pooling",
+			"select get_lock('foo', 10) from dual",
+			"SelectLockFunc not allowed without reserved connection",
+		}, {
+			"setting system variables must happen inside reserved connections",
+			"set sql_safe_updates = false",
+			"Set not allowed without reserved connection",
+		}, {
+			"setting system variables must happen inside reserved connections",
+			"set @@sql_safe_updates = false",
+			"Set not allowed without reserved connection",
+		}, {
+			"setting system variables must happen inside reserved connections",
+			"set @udv = false",
+			"Set not allowed without reserved connection",
+		},
+	}
+	for _, tcase := range tcases {
+		t.Run(tcase.name, func(t *testing.T) {
+			statement, err := sqlparser.Parse(tcase.query)
+			require.NoError(t, err)
+			plan, err := planbuilder.Build(statement, map[string]*schema.Table{}, "dbName")
+			// Plan building will not fail, but it will mark that reserved connection is needed.
+			// checking plan is valid will fail.
+			require.NoError(t, err)
+			require.True(t, plan.NeedsReservedConn)
+			err = isValid(plan.PlanID, false, false)
+			require.EqualError(t, err, tcase.err)
+		})
+	}
+}