Repository: vitessio/vitess
Pull request #10291 (2022-05-12 15:21:00)

Improve the formatting for partitioning
Right now this is all crammed onto a single line, but these can be quite extensive.

With the changes here it now better matches the output of what MySQL also does with formatting partitioning specs with a new line for each partition defined.

## Related Issue(s)

Part of #10203

## Checklist

-   [ ] "Backport me!" label has been added if this change should be backported
-   [x] Tests were added or are not required
-   [x] Documentation was added or is not required

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

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

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

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

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

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

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

Patch:
diff --git a/.github/workflows/check_make_vtadmin_web_proto.yml b/.github/workflows/check_make_vtadmin_web_proto.yml
new file mode 100644
index 0000000000..c69b72a362
--- /dev/null
+++ b/.github/workflows/check_make_vtadmin_web_proto.yml
@@ -0,0 +1,49 @@
+name: check_make_vtadmin_web_proto
+on: [push, pull_request]
+jobs:
+
+  build:
+    name: Check Make VTAdmin Web Proto
+    runs-on: ubuntu-latest
+    steps:
+    - name: Check out code
+      uses: actions/checkout@v2
+
+    - name: Check for changes in relevant files
+      uses: frouioui/paths-filter@main
+      id: changes
+      with:
+        token: ''
+        filters: |
+          proto_changes:
+            - 'bootstrap.sh'
+            - 'tools/**'
+            - 'build.env'
+            - 'go.[sumod]'
+            - 'Makefile'
+            - 'go/vt/proto/**'
+            - 'proto/*.proto'
+            - 'web/vtadmin/src/proto/**'
+
+    - name: Set up Go
+      uses: actions/setup-go@v2
+      if: steps.changes.outputs.proto_changes == 'true'
+      with:
+        go-version: 1.18.1
+
+    - name: Setup Node
+      if: steps.changes.outputs.proto_changes == 'true'
+      uses: actions/setup-node@v2
+      with:
+        # node-version should match package.json
+        node-version: '16.13.0'
+
+    - name: Install npm dependencies
+      if: steps.changes.outputs.proto_changes == 'true'
+      run: npm ci
+      working-directory: ./web/vtadmin
+
+    - name: check_make_vtadmin_web_proto
+      if: steps.changes.outputs.proto_changes == 'true'
+      run: |
+        tools/check_make_vtadmin_web_proto.sh
diff --git a/go/vt/sqlparser/ast_format.go b/go/vt/sqlparser/ast_format.go
index c1c4bafecf..28c9080baf 100644
--- a/go/vt/sqlparser/ast_format.go
+++ b/go/vt/sqlparser/ast_format.go
@@ -543,7 +543,7 @@ func (node *PartitionEngine) Format(buf *TrackedBuffer) {
 
 // Format formats the node.
 func (node *PartitionOption) Format(buf *TrackedBuffer) {
-	buf.literal("partition by")
+	buf.literal("\npartition by")
 	if node.IsLinear {
 		buf.literal(" linear")
 	}
@@ -573,10 +573,10 @@ func (node *PartitionOption) Format(buf *TrackedBuffer) {
 		buf.astPrintf(node, " %v", node.SubPartition)
 	}
 	if node.Definitions != nil {
-		buf.literal(" (")
+		buf.literal("\n(")
 		for i, pd := range node.Definitions {
 			if i != 0 {
-				buf.literal(", ")
+				buf.literal(",\n ")
 			}
 			buf.astPrintf(node, "%v", pd)
 		}
@@ -643,7 +643,7 @@ func (ts *TableSpec) Format(buf *TrackedBuffer) {
 		}
 	}
 	if ts.PartitionOption != nil {
-		buf.astPrintf(ts, " %v", ts.PartitionOption)
+		buf.astPrintf(ts, "%v", ts.PartitionOption)
 	}
 }
 
diff --git a/go/vt/sqlparser/ast_format_fast.go b/go/vt/sqlparser/ast_format_fast.go
index a68a6de682..f1206ba0c0 100644
--- a/go/vt/sqlparser/ast_format_fast.go
+++ b/go/vt/sqlparser/ast_format_fast.go
@@ -732,7 +732,7 @@ func (node *PartitionEngine) formatFast(buf *TrackedBuffer) {
 
 // formatFast formats the node.
 func (node *PartitionOption) formatFast(buf *TrackedBuffer) {
-	buf.WriteString("partition by")
+	buf.WriteString("\npartition by")
 	if node.IsLinear {
 		buf.WriteString(" linear")
 	}
@@ -772,10 +772,10 @@ func (node *PartitionOption) formatFast(buf *TrackedBuffer) {
 		node.SubPartition.formatFast(buf)
 	}
 	if node.Definitions != nil {
-		buf.WriteString(" (")
+		buf.WriteString("\n(")
 		for i, pd := range node.Definitions {
 			if i != 0 {
-				buf.WriteString(", ")
+				buf.WriteString(",\n ")
 			}
 			pd.formatFast(buf)
 		}
@@ -857,7 +857,6 @@ func (ts *TableSpec) formatFast(buf *TrackedBuffer) {
 		}
 	}
 	if ts.PartitionOption != nil {
-		buf.WriteByte(' ')
 		ts.PartitionOption.formatFast(buf)
 	}
 }
diff --git a/go/vt/sqlparser/ast_funcs.go b/go/vt/sqlparser/ast_funcs.go
index 374d595f68..966893730f 100644
--- a/go/vt/sqlparser/ast_funcs.go
+++ b/go/vt/sqlparser/ast_funcs.go
@@ -1813,6 +1813,19 @@ func (es *ExtractedSubquery) updateAlternative() {
 	}
 }
 
+// ColumnName returns the alias if one was provided, otherwise prints the AST
+func (ae *AliasedExpr) ColumnName() string {
+	if !ae.As.IsEmpty() {
+		return ae.As.String()
+	}
+
+	if col, ok := ae.Expr.(*ColName); ok {
+		return col.Name.String()
+	}
+
+	return String(ae.Expr)
+}
+
 func isExprLiteral(expr Expr) bool {
 	switch expr := expr.(type) {
 	case *Literal:
diff --git a/go/vt/vtadmin/cluster/cluster.go b/go/vt/vtadmin/cluster/cluster.go
index cf5796d1de..4294af9f37 100644
--- a/go/vt/vtadmin/cluster/cluster.go
+++ b/go/vt/vtadmin/cluster/cluster.go
@@ -785,7 +785,12 @@ func (c *Cluster) GetCellInfos(ctx context.Context, req *vtadminpb.GetCellInfosR
 
 	names := req.Cells
 	if len(names) == 0 {
+		if err := c.topoReadPool.Acquire(ctx); err != nil {
+			return nil, fmt.Errorf("GetCellInfoNames() failed to acquire topoReadPool: %w", err)
+		}
 		resp, err := c.Vtctld.GetCellInfoNames(ctx, &vtctldatapb.GetCellInfoNamesRequest{})
+		c.topoReadPool.Release()
+
 		if err != nil {
 			return nil, fmt.Errorf("failed to GetCellInfoNames: %w", err)
 		}
@@ -826,9 +831,15 @@ func (c *Cluster) GetCellInfos(ctx context.Context, req *vtadminpb.GetCellInfosR
 		go func(name string) {
 			defer wg.Done()
 
+			if err := c.topoReadPool.Acquire(ctx); err != nil {
+				rec.RecordError(fmt.Errorf("GetCellInfo(%s) failed to acquire topoReadPool: %w", name, err))
+				return
+			}
 			resp, err := c.Vtctld.GetCellInfo(ctx, &vtctldatapb.GetCellInfoRequest{
 				Cell: name,
 			})
+			c.topoReadPool.Release()
+
 			if err != nil {
 				rec.RecordError(fmt.Errorf("GetCellInfo(%s) failed: %w", name, err))
 				return
@@ -857,6 +868,11 @@ func (c *Cluster) GetCellsAliases(ctx context.Context) (*vtadminpb.ClusterCellsA
 	span, ctx := trace.NewSpan(ctx, "Cluster.GetCellsAliases")
 	defer span.Finish()
 
+	if err := c.topoReadPool.Acquire(ctx); err != nil {
+		return nil, fmt.Errorf("GetCellsAliases() failed to acquire topoReadPool: %w", err)
+	}
+	defer c.topoReadPool.Release()
+
 	resp, err := c.Vtctld.GetCellsAliases(ctx, &vtctldatapb.GetCellsAliasesRequest{})
 	if err != nil {
 		return nil, err
diff --git a/go/vt/vtgate/planbuilder/abstract/queryprojection.go b/go/vt/vtgate/planbuilder/abstract/queryprojection.go
index f78fda310e..b80d23c744 100644
--- a/go/vt/vtgate/planbuilder/abstract/queryprojection.go
+++ b/go/vt/vtgate/planbuilder/abstract/queryprojection.go
@@ -444,19 +444,12 @@ func (qp *QueryProjection) AggregationExpressions() (out []Aggr, err error) {
 			}
 		}
 
-		var alias string
-		if aliasedExpr.As.IsEmpty() {
-			alias = sqlparser.String(aliasedExpr.Expr)
-		} else {
-			alias = aliasedExpr.As.String()
-		}
-
 		idxCopy := idx
 		out = append(out, Aggr{
 			Original: aliasedExpr,
 			Func:     fExpr,
 			OpCode:   opcode,
-			Alias:    alias,
+			Alias:    aliasedExpr.ColumnName(),
 			Index:    &idxCopy,
 			Distinct: fExpr.Distinct,
 		})
diff --git a/go/vt/vtgate/planbuilder/horizon_planning.go b/go/vt/vtgate/planbuilder/horizon_planning.go
index 7831c13994..0033360542 100644
--- a/go/vt/vtgate/planbuilder/horizon_planning.go
+++ b/go/vt/vtgate/planbuilder/horizon_planning.go
@@ -604,7 +604,8 @@ func passGroupingColumns(proj *projection, groupings []offsets, grouping []abstr
 	for idx, grp := range groupings {
 		origGrp := grouping[idx]
 		var offs offsets
-		offs.col, err = proj.addColumn(origGrp.InnerIndex, sqlparser.Offset(grp.col), "")
+		alias := origGrp.AsAliasedExpr().ColumnName()
+		offs.col, err = proj.addColumn(origGrp.InnerIndex, sqlparser.Offset(grp.col), alias)
 		if err != nil {
 			return nil, err
 		}
diff --git a/go/vt/vtgate/planbuilder/ordered_aggregate.go b/go/vt/vtgate/planbuilder/ordered_aggregate.go
index 4c091055dd..c79120a8a8 100644
--- a/go/vt/vtgate/planbuilder/ordered_aggregate.go
+++ b/go/vt/vtgate/planbuilder/ordered_aggregate.go
@@ -285,12 +285,6 @@ func (oa *orderedAggregate) pushAggr(pb *primitiveBuilder, expr *sqlparser.Alias
 		}
 		oa.extraDistinct = col
 		oa.preProcess = true
-		var alias string
-		if expr.As.IsEmpty() {
-			alias = sqlparser.String(expr.Expr)
-		} else {
-			alias = expr.As.String()
-		}
 		switch opcode {
 		case engine.AggregateCount:
 			opcode = engine.AggregateCountDistinct
@@ -300,7 +294,7 @@ func (oa *orderedAggregate) pushAggr(pb *primitiveBuilder, expr *sqlparser.Alias
 		oa.aggregates = append(oa.aggregates, &engine.AggregateParams{
 			Opcode: opcode,
 			Col:    innerCol,
-			Alias:  alias,
+			Alias:  expr.ColumnName(),
 		})
 	} else {
 		newBuilder, _, innerCol, err := planProjection(pb, oa.input, expr, origin)
diff --git a/go/vt/vtgate/planbuilder/vindex_func.go b/go/vt/vtgate/planbuilder/vindex_func.go
index 5b951ec40d..fbc755baac 100644
--- a/go/vt/vtgate/planbuilder/vindex_func.go
+++ b/go/vt/vtgate/planbuilder/vindex_func.go
@@ -161,15 +161,8 @@ func (vf *vindexFunc) SupplyProjection(expr *sqlparser.AliasedExpr, reuse bool)
 		}
 	}
 
-	var name string
-	if expr.As.IsEmpty() {
-		name = sqlparser.String(colName)
-	} else {
-		name = expr.As.String()
-	}
-
 	vf.eVindexFunc.Fields = append(vf.eVindexFunc.Fields, &querypb.Field{
-		Name: name,
+		Name: expr.ColumnName(),
 		Type: querypb.Type_VARBINARY,
 	})
 	vf.eVindexFunc.Cols = append(vf.eVindexFunc.Cols, enum)
diff --git a/go/vt/vtgate/semantics/analyzer.go b/go/vt/vtgate/semantics/analyzer.go
index 0ad2d3cb1d..b2a8cddecc 100644
--- a/go/vt/vtgate/semantics/analyzer.go
+++ b/go/vt/vtgate/semantics/analyzer.go
@@ -53,9 +53,11 @@ func newAnalyzer(dbName string, si SchemaInformation) *analyzer {
 	}
 	s.org = a
 	a.tables.org = a
-	a.binder = newBinder(s, a, a.tables, a.typer)
-	a.rewriter = &earlyRewriter{scoper: s}
 
+	b := newBinder(s, a, a.tables, a.typer)
+	a.binder = b
+	a.rewriter = &earlyRewriter{scoper: s, binder: b}
+	s.binder = b
 	return a
 }
 
@@ -302,9 +304,6 @@ func (a *analyzer) checkForInvalidConstructs(cursor *sqlparser.Cursor) error {
 			return vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "NEXT used on a non-sequence table")
 		}
 	case *sqlparser.JoinTableExpr:
-		if node.Condition != nil && node.Condition.Using != nil {
-			return UnshardedError{Inner: vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: join with USING(column_list) clause for complex queries")}
-		}
 		if node.Join == sqlparser.NaturalJoinType || node.Join == sqlparser.NaturalRightJoinType || node.Join == sqlparser.NaturalLeftJoinType {
 			return vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: "+node.Join.ToString())
 		}
diff --git a/go/vt/vtgate/semantics/binder.go b/go/vt/vtgate/semantics/binder.go
index 13c06709fd..d3d7ab7af9 100644
--- a/go/vt/vtgate/semantics/binder.go
+++ b/go/vt/vtgate/semantics/binder.go
@@ -17,6 +17,8 @@ limitations under the License.
 package semantics
 
 import (
+	"strings"
+
 	"vitess.io/vitess/go/vt/vtgate/engine"
 
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
@@ -37,18 +39,24 @@ type binder struct {
 	typer       *typer
 	subqueryMap map[sqlparser.Statement][]*sqlparser.ExtractedSubquery
 	subqueryRef map[*sqlparser.Subquery]*sqlparser.ExtractedSubquery
+
+	// every table will have an entry in the outer map. it will point to a map with all the columns
+	// that this map is joined with using USING.
+	// This information is used to expand `*` correctly, and is not available post-analysis
+	usingJoinInfo map[TableSet]map[string]TableSet
 }
 
 func newBinder(scoper *scoper, org originable, tc *tableCollector, typer *typer) *binder {
 	return &binder{
-		recursive:   map[sqlparser.Expr]TableSet{},
-		direct:      map[sqlparser.Expr]TableSet{},
-		scoper:      scoper,
-		org:         org,
-		tc:          tc,
-		typer:       typer,
-		subqueryMap: map[sqlparser.Statement][]*sqlparser.ExtractedSubquery{},
-		subqueryRef: map[*sqlparser.Subquery]*sqlparser.ExtractedSubquery{},
+		recursive:     map[sqlparser.Expr]TableSet{},
+		direct:        map[sqlparser.Expr]TableSet{},
+		scoper:        scoper,
+		org:           org,
+		tc:            tc,
+		typer:         typer,
+		subqueryMap:   map[sqlparser.Statement][]*sqlparser.ExtractedSubquery{},
+		subqueryRef:   map[*sqlparser.Subquery]*sqlparser.ExtractedSubquery{},
+		usingJoinInfo: map[TableSet]map[string]TableSet{},
 	}
 }
 
@@ -65,11 +73,40 @@ func (b *binder) up(cursor *sqlparser.Cursor) error {
 		b.subqueryRef[node] = sq
 
 		b.setSubQueryDependencies(node, currScope)
-
+	case *sqlparser.JoinCondition:
+		currScope := b.scoper.currentScope()
+		for _, ident := range node.Using {
+			name := sqlparser.NewColName(ident.String())
+			deps, err := b.resolveColumn(name, currScope, true)
+			if err != nil {
+				return err
+			}
+			currScope.joinUsing[ident.Lowered()] = deps.direct
+		}
+		if len(node.Using) > 0 {
+			err := rewriteJoinUsing(currScope, node.Using, b.org)
+			if err != nil {
+				return err
+			}
+			node.Using = nil
+		}
 	case *sqlparser.ColName:
-		deps, err := b.resolveColumn(node, b.scoper.currentScope())
+		currentScope := b.scoper.currentScope()
+		deps, err := b.resolveColumn(node, currentScope, false)
 		if err != nil {
-			return err
+			if deps.direct.NumberOfTables() == 0 ||
+				!strings.HasSuffix(err.Error(), "is ambiguous") ||
+				!b.canRewriteUsingJoin(deps, node) {
+				return err
+			}
+
+			// if we got here it means we are dealing with a ColName that is involved in a JOIN USING.
+			// we do the rewriting of these ColName structs here because it would be difficult to copy all the
+			// needed state over to the earlyRewriter
+			deps, err = b.rewriteJoinUsingColName(deps, node, currentScope)
+			if err != nil {
+				return err
+			}
 		}
 		b.recursive[node] = deps.recursive
 		b.direct[node] = deps.direct
@@ -98,6 +135,48 @@ func (b *binder) up(cursor *sqlparser.Cursor) error {
 	return nil
 }
 
+func (b *binder) rewriteJoinUsingColName(deps dependency, node *sqlparser.ColName, currentScope *scope) (dependency, error) {
+	constituents := deps.recursive.Constituents()
+	if len(constituents) < 1 {
+		return dependency{}, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "not expected - we should not a colname that depends on nothing")
+	}
+	newTbl := constituents[0]
+	infoFor, err := b.tc.tableInfoFor(newTbl)
+	if err != nil {
+		return dependency{}, err
+	}
+	alias := infoFor.getExpr().As
+	if alias.IsEmpty() {
+		name, err := infoFor.Name()
+		if err != nil {
+			return dependency{}, err
+		}
+		node.Qualifier = name
+	} else {
+		node.Qualifier = sqlparser.TableName{
+			Name: sqlparser.NewTableIdent(alias.String()),
+		}
+	}
+	deps, err = b.resolveColumn(node, currentScope, false)
+	if err != nil {
+		return dependency{}, err
+	}
+	return deps, nil
+}
+
+// canRewriteUsingJoin will return true when this ColName is safe to rewrite since it can only belong to a USING JOIN
+func (b *binder) canRewriteUsingJoin(deps dependency, node *sqlparser.ColName) bool {
+	tbls := deps.direct.Constituents()
+	colName := node.Name.Lowered()
+	for _, tbl := range tbls {
+		m := b.usingJoinInfo[tbl]
+		if _, found := m[colName]; !found {
+			return false
+		}
+	}
+	return true
+}
+
 // setSubQueryDependencies sets the correct dependencies for the subquery
 // the binder usually only sets the dependencies of ColNames, but we need to
 // handle the subquery dependencies differently, so they are set manually here
@@ -154,25 +233,32 @@ func (b *binder) createExtractedSubquery(cursor *sqlparser.Cursor, currScope *sc
 	return sq, nil
 }
 
-func (b *binder) resolveColumn(colName *sqlparser.ColName, current *scope) (deps dependency, err error) {
+func (b *binder) resolveColumn(colName *sqlparser.ColName, current *scope, allowMulti bool) (dependency, error) {
 	var thisDeps dependencies
 	for current != nil {
-		thisDeps, err = b.resolveColumnInScope(current, colName)
+		var err error
+		thisDeps, err = b.resolveColumnInScope(current, colName, allowMulti)
 		if err != nil {
 			err = makeAmbiguousError(colName, err)
-			return dependency{}, err
+			if thisDeps == nil {
+				return dependency{}, err
+			}
 		}
 		if !thisDeps.empty() {
-			deps, err = thisDeps.get()
-			err = makeAmbiguousError(colName, err)
+			deps, thisErr := thisDeps.get()
+			if thisErr != nil {
+				err = makeAmbiguousError(colName, thisErr)
+			}
 			return deps, err
+		} else if err != nil {
+			return dependency{}, err
 		}
 		current = current.parent
 	}
 	return dependency{}, vterrors.NewErrorf(vtrpcpb.Code_INVALID_ARGUMENT, vterrors.BadFieldError, "symbol %s not found", sqlparser.String(colName))
 }
 
-func (b *binder) resolveColumnInScope(current *scope, expr *sqlparser.ColName) (dependencies, error) {
+func (b *binder) resolveColumnInScope(current *scope, expr *sqlparser.ColName, allowMulti bool) (dependencies, error) {
 	var deps dependencies = &nothing{}
 	for _, table := range current.tables {
 		if !expr.Qualifier.IsEmpty() && !table.matches(expr.Qualifier) {
@@ -182,10 +268,7 @@ func (b *binder) resolveColumnInScope(current *scope, expr *sqlparser.ColName) (
 		if err != nil {
 			return nil, err
 		}
-		deps, err = thisDeps.merge(deps)
-		if err != nil {
-			return nil, err
-		}
+		deps = thisDeps.merge(deps, allowMulti)
 	}
 	if deps, isUncertain := deps.(*uncertain); isUncertain && deps.fail {
 		// if we have a failure from uncertain, we matched the column to multiple non-authoritative tables
diff --git a/go/vt/vtgate/semantics/dependencies.go b/go/vt/vtgate/semantics/dependencies.go
index 8d3ade1d09..197f2e95d8 100644
--- a/go/vt/vtgate/semantics/dependencies.go
+++ b/go/vt/vtgate/semantics/dependencies.go
@@ -28,7 +28,7 @@ type (
 	dependencies interface {
 		empty() bool
 		get() (dependency, error)
-		merge(dependencies) (dependencies, error)
+		merge(other dependencies, allowMulti bool) dependencies
 	}
 	dependency struct {
 		direct    TableSet
@@ -38,6 +38,7 @@ type (
 	nothing struct{}
 	certain struct {
 		dependency
+		err error
 	}
 	uncertain struct {
 		dependency
@@ -84,20 +85,18 @@ func (u *uncertain) get() (dependency, error) {
 	return u.dependency, nil
 }
 
-func (u *uncertain) merge(d dependencies) (dependencies, error) {
+func (u *uncertain) merge(d dependencies, _ bool) dependencies {
 	switch d := d.(type) {
-	case *nothing:
-		return u, nil
 	case *uncertain:
-		if d.recursive == u.recursive {
-			return u, nil
+		if d.recursive != u.recursive {
+			u.fail = true
 		}
-		u.fail = true
-		return u, nil
+		return u
 	case *certain:
-		return d, nil
+		return d
+	default:
+		return u
 	}
-	return nil, ambigousErr
 }
 
 func (c *certain) empty() bool {
@@ -105,20 +104,25 @@ func (c *certain) empty() bool {
 }
 
 func (c *certain) get() (dependency, error) {
-	return c.dependency, nil
+	return c.dependency, c.err
 }
 
-func (c *certain) merge(d dependencies) (dependencies, error) {
+func (c *certain) merge(d dependencies, allowMulti bool) dependencies {
 	switch d := d.(type) {
-	case *nothing, *uncertain:
-		return c, nil
 	case *certain:
 		if d.recursive == c.recursive {
-			return c, nil
+			return c
 		}
+		c.direct.MergeInPlace(d.direct)
+		c.recursive.MergeInPlace(d.recursive)
+		if !allowMulti {
+			c.err = ambigousErr
+		}
+
+		return c
 	}
 
-	return nil, ambigousErr
+	return c
 }
 
 func (n *nothing) empty() bool {
@@ -129,6 +133,6 @@ func (n *nothing) get() (dependency, error) {
 	return dependency{}, nil
 }
 
-func (n *nothing) merge(d dependencies) (dependencies, error) {
-	return d, nil
+func (n *nothing) merge(d dependencies, _ bool) dependencies {
+	return d
 }
diff --git a/go/vt/vtgate/semantics/early_rewriter.go b/go/vt/vtgate/semantics/early_rewriter.go
index f8ef08953b..dc88ccec2e 100644
--- a/go/vt/vtgate/semantics/early_rewriter.go
+++ b/go/vt/vtgate/semantics/early_rewriter.go
@@ -25,6 +25,7 @@ import (
 )
 
 type earlyRewriter struct {
+	binder  *binder
 	scoper  *scoper
 	clause  string
 	warning string
@@ -37,7 +38,7 @@ func (r *earlyRewriter) down(cursor *sqlparser.Cursor) error {
 		if !isSel {
 			return nil
 		}
-		tables := r.scoper.currentScope().tables
+		currentScope := r.scoper.currentScope()
 		var selExprs sqlparser.SelectExprs
 		changed := false
 		for _, selectExpr := range node {
@@ -46,7 +47,7 @@ func (r *earlyRewriter) down(cursor *sqlparser.Cursor) error {
 				selExprs = append(selExprs, selectExpr)
 				continue
 			}
-			starExpanded, colNames, err := expandTableColumns(tables, starExpr)
+			starExpanded, colNames, err := expandTableColumns(starExpr, currentScope.tables, r.binder.usingJoinInfo, r.scoper.org)
 			if err != nil {
 				return err
 			}
@@ -149,7 +150,75 @@ func realCloneOfColNames(expr sqlparser.Expr, union bool) sqlparser.Expr {
 	}, nil).(sqlparser.Expr)
 }
 
-func expandTableColumns(tables []TableInfo, starExpr *sqlparser.StarExpr) (bool, sqlparser.SelectExprs, error) {
+func rewriteJoinUsing(
+	current *scope,
+	using sqlparser.Columns,
+	org originable,
+) error {
+	joinUsing := current.prepareUsingMap()
+	predicates := make([]sqlparser.Expr, 0, len(using))
+	for _, column := range using {
+		var foundTables []sqlparser.TableName
+		for _, tbl := range current.tables {
+			if !tbl.authoritative() {
+				return vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "can't handle JOIN USING without authoritative tables")
+			}
+
+			currTable := tbl.getTableSet(org)
+			usingCols := joinUsing[currTable]
+			if usingCols == nil {
+				usingCols = map[string]TableSet{}
+			}
+			for _, col := range tbl.getColumns() {
+				_, found := usingCols[col.Name]
+				if found {
+					tblName, err := tbl.Name()
+					if err != nil {
+						return err
+					}
+
+					foundTables = append(foundTables, tblName)
+					break // no need to look at other columns in this table
+				}
+			}
+		}
+		for i, lft := range foundTables {
+			for j := i + 1; j < len(foundTables); j++ {
+				rgt := foundTables[j]
+				predicates = append(predicates, &sqlparser.ComparisonExpr{
+					Operator: sqlparser.EqualOp,
+					Left:     sqlparser.NewColNameWithQualifier(column.String(), lft),
+					Right:    sqlparser.NewColNameWithQualifier(column.String(), rgt),
+				})
+			}
+		}
+	}
+
+	// now, we go up the scope until we find a SELECT with a where clause we can add this predicate to
+	for current != nil {
+		sel, found := current.stmt.(*sqlparser.Select)
+		if found {
+			if sel.Where == nil {
+				sel.Where = &sqlparser.Where{
+					Type: sqlparser.WhereClause,
+					Expr: sqlparser.AndExpressions(predicates...),
+				}
+			} else {
+				sel.Where.Expr = sqlparser.AndExpressions(append(predicates, sel.Where.Expr)...)
+			}
+			return nil
+		}
+		current = current.parent
+	}
+	return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "did not find WHERE clause")
+}
+
+func expandTableColumns(
+	starExpr *sqlparser.StarExpr,
+	tables []TableInfo,
+	joinUsing map[TableSet]map[string]TableSet,
+	org originable,
+) (bool, sqlparser.SelectExprs, error) {
 	unknownTbl := true
 	var colNames sqlparser.SelectExprs
 	starExpanded := true
@@ -169,7 +238,13 @@ func expandTableColumns(tables []TableInfo, starExpr *sqlparser.StarExpr) (bool,
 
 		withAlias := len(tables) > 1
 		withQualifier := withAlias || !tbl.getExpr().As.IsEmpty()
-		for _, col := range tbl.getColumns() {
+		currTable := tbl.getTableSet(org)
+		usingCols := joinUsing[currTable]
+		if usingCols == nil {
+			usingCols = map[string]TableSet{}
+		}
+
+		addColName := func(col ColumnInfo) {
 			var colName *sqlparser.ColName
 			var alias sqlparser.ColIdent
 			if withQualifier {
@@ -182,6 +257,40 @@ func expandTableColumns(tables []TableInfo, starExpr *sqlparser.StarExpr) (bool,
 			}
 			colNames = append(colNames, &sqlparser.AliasedExpr{Expr: colName, As: alias})
 		}
+
+		/*
+			Redundant column elimination and column ordering occurs according to standard SQL, producing this display order:
+			  *	First, coalesced common columns of the two joined tables, in the order in which they occur in the first table
+			  *	Second, columns unique to the first table, in order in which they occur in that table
+			  *	Third, columns unique to the second table, in order in which they occur in that table
+
+			From: https://dev.mysql.com/doc/refman/8.0/en/join.html
+		*/
+	outer:
+		// in this first loop we just find columns used in any JOIN USING used on this table
+		for _, col := range tbl.getColumns() {
+			ts, found := usingCols[col.Name]
+			if found {
+				for i, ts := range ts.Constituents() {
+					if ts.Equals(currTable) {
+						if i == 0 {
+							addColName(col)
+						} else {
+							continue outer
+						}
+					}
+				}
+			}
+		}
+
+		// and this time around we are printing any columns not involved in any JOIN USING
+		for _, col := range tbl.getColumns() {
+			if ts, found := usingCols[col.Name]; found && currTable.IsSolvedBy(ts) {
+				continue
+			}
+
+			addColName(col)
+		}
 	}
 
 	if unknownTbl {
diff --git a/go/vt/vtgate/semantics/scoper.go b/go/vt/vtgate/semantics/scoper.go
index 1d70829bed..97ec5f38b7 100644
--- a/go/vt/vtgate/semantics/scoper.go
+++ b/go/vt/vtgate/semantics/scoper.go
@@ -34,16 +34,18 @@ type (
 		wScope map[*sqlparser.Select]*scope
 		scopes []*scope
 		org    originable
+		binder *binder
 
 		// These scopes are only used for rewriting ORDER BY 1 and GROUP BY 1
 		specialExprScopes map[*sqlparser.Literal]*scope
 	}
 
 	scope struct {
-		parent  *scope
-		stmt    sqlparser.Statement
-		tables  []TableInfo
-		isUnion bool
+		parent    *scope
+		stmt      sqlparser.Statement
+		tables    []TableInfo
+		isUnion   bool
+		joinUsing map[string]TableSet
 	}
 )
 
@@ -238,12 +240,19 @@ func (s *scoper) push(sc *scope) {
 }
 
 func (s *scoper) popScope() {
+	usingMap := s.currentScope().prepareUsingMap()
+	for ts, m := range usingMap {
+		s.binder.usingJoinInfo[ts] = m
+	}
 	l := len(s.scopes) - 1
 	s.scopes = s.scopes[:l]
 }
 
 func newScope(parent *scope) *scope {
-	return &scope{parent: parent}
+	return &scope{
+		parent:    parent,
+		joinUsing: map[string]TableSet{},
+	}
 }
 
 func (s *scope) addTable(info TableInfo) error {
@@ -265,3 +274,18 @@ func (s *scope) addTable(info TableInfo) error {
 	s.tables = append(s.tables, info)
 	return nil
 }
+
+func (s *scope) prepareUsingMap() (result map[TableSet]map[string]TableSet) {
+	result = map[TableSet]map[string]TableSet{}
+	for colName, tss := range s.joinUsing {
+		for _, ts := range tss.Constituents() {
+			m := result[ts]
+			if m == nil {
+				m = map[string]TableSet{}
+			}
+			m[colName] = tss
+			result[ts] = m
+		}
+	}
+	return
+}
diff --git a/go/vt/vtgate/semantics/table_collector.go b/go/vt/vtgate/semantics/table_collector.go
index 9af71cad99..181e949610 100644
--- a/go/vt/vtgate/semantics/table_collector.go
+++ b/go/vt/vtgate/semantics/table_collector.go
@@ -134,6 +134,15 @@ func (tc *tableCollector) tableSetFor(t *sqlparser.AliasedTableExpr) TableSet {
 	panic("unknown table")
 }
 
+// tableInfoFor returns the table info for the table set. It should contains only single table.
+func (tc *tableCollector) tableInfoFor(id TableSet) (TableInfo, error) {
+	offset := id.TableOffset()
+	if offset < 0 {
+		return nil, ErrMultipleTables
+	}
+	return tc.Tables[offset], nil
+}
+
 func (tc *tableCollector) createTable(
 	t sqlparser.TableName,
 	alias *sqlparser.AliasedTableExpr,
diff --git a/go/vt/vtgate/semantics/tabletset.go b/go/vt/vtgate/semantics/tabletset.go
index 3a0290a280..422579d729 100644
--- a/go/vt/vtgate/semantics/tabletset.go
+++ b/go/vt/vtgate/semantics/tabletset.go
@@ -254,9 +254,9 @@ func (ts TableSet) ForEachTable(callback func(int)) {
 }
 
 // Constituents returns a slice with the indices for all tables in this TableSet
-func (ts TableSet) Constituents() (result []int) {
+func (ts TableSet) Constituents() (result []TableSet) {
 	ts.ForEachTable(func(t int) {
-		result = append(result, t)
+		result = append(result, SingleTableSet(t))
 	})
 	return
 }
diff --git a/go/vt/vtgate/semantics/vtable.go b/go/vt/vtgate/semantics/vtable.go
index c992246e5d..50d13d5719 100644
--- a/go/vt/vtgate/semantics/vtable.go
+++ b/go/vt/vtgate/semantics/vtable.go
@@ -37,7 +37,6 @@ var _ TableInfo = (*vTableInfo)(nil)
 // dependencies implements the TableInfo interface
 func (v *vTableInfo) dependencies(colName string, org originable) (dependencies, error) {
 	var deps dependencies = &nothing{}
-	var err error
 	for i, name := range v.columnNames {
 		if name != colName {
 			continue
@@ -45,10 +44,7 @@ func (v *vTableInfo) dependencies(colName string, org originable) (dependencies,
 		directDeps, recursiveDeps, qt := org.depsForExpr(v.cols[i])
 
 		newDeps := createCertain(directDeps, recursiveDeps, qt)
-		deps, err = deps.merge(newDeps)
-		if err != nil {
-			return nil, err
-		}
+		deps = deps.merge(newDeps, false)
 	}
 	if deps.empty() && v.hasStar() {
 		return createUncertain(v.tables, v.tables), nil
@@ -97,7 +93,7 @@ func (v *vTableInfo) hasStar() bool {
 }
 
 // GetTables implements the TableInfo interface
-func (v *vTableInfo) getTableSet(org originable) TableSet {
+func (v *vTableInfo) getTableSet(_ originable) TableSet {
 	return v.tables
 }
 
diff --git a/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go b/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
index 18508f65a8..586084fad9 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/planbuilder.go
@@ -584,10 +584,6 @@ func (plan *Plan) analyzeExpr(vschema *localVSchema, selExpr sqlparser.SelectExp
 		if err != nil {
 			return ColExpr{}, err
 		}
-		as := aliased.As
-		if as.IsEmpty() {
-			as = sqlparser.NewColIdent(sqlparser.String(aliased.Expr))
-		}
 		return ColExpr{
 			ColNum: colnum,
 			Field:  plan.Table.Fields[colnum],
diff --git a/tools/check_make_vtadmin_web_proto.sh b/tools/check_make_vtadmin_web_proto.sh
new file mode 100755
index 0000000000..d7b2a1d657
--- /dev/null
+++ b/tools/check_make_vtadmin_web_proto.sh
@@ -0,0 +1,19 @@
+#!/bin/bash
+
+source build.env
+
+first_output=$(git status --porcelain)
+
+make vtadmin_web_proto_types
+
+second_output=$(git status --porcelain)
+
+diff=$(diff <( echo "$first_output") <( echo "$second_output"))
+
+if [[ "$diff" != "" ]]; then
+  echo "ERROR: Regenerated vtadmin web proto files do not match the current version."
+  echo -e "List of files containing differences:\n$diff"
+  exit 1
+fi
+
+echo "VTAdmin Web Protos are up to date"
\ No newline at end of file
diff --git a/web/vtadmin/src/components/routes/tablet/DangerAction.tsx b/web/vtadmin/src/components/DangerAction.tsx
similarity index 77%
rename from web/vtadmin/src/components/routes/tablet/DangerAction.tsx
rename to web/vtadmin/src/components/DangerAction.tsx
index 20c14a9872..5bbdfd748e 100644
--- a/web/vtadmin/src/components/routes/tablet/DangerAction.tsx
+++ b/web/vtadmin/src/components/DangerAction.tsx
@@ -1,7 +1,23 @@
+/**
+ * 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.
+ */
+
 import React, { useState } from 'react';
 import { UseMutationResult } from 'react-query';
-import { Icon, Icons } from '../../Icon';
-import { TextInput } from '../../TextInput';
+import { Icon, Icons } from './Icon';
+import { TextInput } from './TextInput';
 
 type Mutation = UseMutationResult & {
     mutate: () => void;
diff --git a/web/vtadmin/src/components/routes/tablet/Advanced.tsx b/web/vtadmin/src/components/routes/tablet/Advanced.tsx
index d023454f7d..e7bfd77d2e 100644
--- a/web/vtadmin/src/components/routes/tablet/Advanced.tsx
+++ b/web/vtadmin/src/components/routes/tablet/Advanced.tsx
@@ -1,3 +1,19 @@
+/**
+ * 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.
+ */
+
 import React from 'react';
 import { UseMutationResult } from 'react-query';
 import { useHistory, useParams } from 'react-router-dom';
@@ -11,9 +27,9 @@ import {
 } from '../../../hooks/api';
 import { vtadmin } from '../../../proto/vtadmin';
 import { isPrimary } from '../../../util/tablets';
+import DangerAction from '../../DangerAction';
 import { Icon, Icons } from '../../Icon';
 import { success, warn } from '../../Snackbar';
-import DangerAction from './DangerAction';
 
 interface AdvancedProps {
     tablet: vtadmin.Tablet | undefined;


Test patch:
diff --git a/go/test/endtoend/utils/cmp.go b/go/test/endtoend/utils/cmp.go
index 72a0775b27..70dc42be8a 100644
--- a/go/test/endtoend/utils/cmp.go
+++ b/go/test/endtoend/utils/cmp.go
@@ -89,6 +89,16 @@ func (mcmp *MySQLCompare) AssertMatchesNoOrder(query, expected string) {
 	assert.Equal(mcmp.t, utils.SortString(expected), utils.SortString(actual), "for query: [%s] expected \n%s \nbut actual \n%s", query, expected, actual)
 }
 
+// AssertMatchesNoOrderInclColumnNames executes the given query against both Vitess and MySQL.
+// The test will be marked as failed if there is a mismatch between the two result sets.
+// This method also checks that the column names are the same and in the same order
+func (mcmp *MySQLCompare) AssertMatchesNoOrderInclColumnNames(query, expected string) {
+	mcmp.t.Helper()
+	qr := mcmp.ExecWithColumnCompare(query)
+	actual := fmt.Sprintf("%v", qr.Rows)
+	assert.Equal(mcmp.t, utils.SortString(expected), utils.SortString(actual), "for query: [%s] expected \n%s \nbut actual \n%s", query, expected, actual)
+}
+
 // AssertIsEmpty executes the given query against both Vitess and MySQL and ensures
 // their results match and are empty.
 func (mcmp *MySQLCompare) AssertIsEmpty(query string) {
@@ -122,7 +132,23 @@ func (mcmp *MySQLCompare) Exec(query string) *sqltypes.Result {
 
 	mysqlQr, err := mcmp.MySQLConn.ExecuteFetch(query, 1000, true)
 	require.NoError(mcmp.t, err, "[MySQL Error] for query: "+query)
-	compareVitessAndMySQLResults(mcmp.t, query, vtQr, mysqlQr)
+	compareVitessAndMySQLResults(mcmp.t, query, vtQr, mysqlQr, false)
+	return vtQr
+}
+
+// ExecWithColumnCompare executes the given query against both Vitess and MySQL and compares
+// the two result set. If there is a mismatch, the difference will be printed and the
+// test will fail. If the query produces an error in either Vitess or MySQL, the test
+// will be marked as failed.
+// The result set of Vitess is returned to the caller.
+func (mcmp *MySQLCompare) ExecWithColumnCompare(query string) *sqltypes.Result {
+	mcmp.t.Helper()
+	vtQr, err := mcmp.VtConn.ExecuteFetch(query, 1000, true)
+	require.NoError(mcmp.t, err, "[Vitess Error] for query: "+query)
+
+	mysqlQr, err := mcmp.MySQLConn.ExecuteFetch(query, 1000, true)
+	require.NoError(mcmp.t, err, "[MySQL Error] for query: "+query)
+	compareVitessAndMySQLResults(mcmp.t, query, vtQr, mysqlQr, true)
 	return vtQr
 }
 
@@ -140,7 +166,7 @@ func (mcmp *MySQLCompare) ExecAllowAndCompareError(query string) (*sqltypes.Resu
 	// Since we allow errors, we don't want to compare results if one of the client failed.
 	// Vitess and MySQL should always be agreeing whether the query returns an error or not.
 	if vtErr == nil && mysqlErr == nil {
-		compareVitessAndMySQLResults(mcmp.t, query, vtQr, mysqlQr)
+		compareVitessAndMySQLResults(mcmp.t, query, vtQr, mysqlQr, false)
 	}
 	return vtQr, vtErr
 }
diff --git a/go/test/endtoend/utils/mysql.go b/go/test/endtoend/utils/mysql.go
index 92ba48cca6..22b6d7ddc0 100644
--- a/go/test/endtoend/utils/mysql.go
+++ b/go/test/endtoend/utils/mysql.go
@@ -23,6 +23,8 @@ import (
 	"path"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/sqlparser"
 
@@ -130,7 +132,7 @@ func prepareMySQLWithSchema(err error, params mysql.ConnParams, sql string) erro
 	return nil
 }
 
-func compareVitessAndMySQLResults(t *testing.T, query string, vtQr *sqltypes.Result, mysqlQr *sqltypes.Result) {
+func compareVitessAndMySQLResults(t *testing.T, query string, vtQr, mysqlQr *sqltypes.Result, compareColumns bool) {
 	if vtQr == nil && mysqlQr == nil {
 		return
 	}
@@ -142,6 +144,23 @@ func compareVitessAndMySQLResults(t *testing.T, query string, vtQr *sqltypes.Res
 		t.Error("MySQL result is 'nil' while Vitess' is not.")
 		return
 	}
+	if compareColumns {
+		vtColCount := len(vtQr.Fields)
+		myColCount := len(mysqlQr.Fields)
+		if vtColCount > 0 && myColCount > 0 {
+			if vtColCount != myColCount {
+				t.Errorf("column count does not match: %d vs %d", vtColCount, myColCount)
+			}
+
+			var vtCols []string
+			var myCols []string
+			for i, vtField := range vtQr.Fields {
+				vtCols = append(vtCols, vtField.Name)
+				myCols = append(myCols, mysqlQr.Fields[i].Name)
+			}
+			assert.Equal(t, myCols, vtCols, "column names do not match - the expected values are what mysql produced")
+		}
+	}
 	stmt, err := sqlparser.Parse(query)
 	if err != nil {
 		t.Error(err)
diff --git a/go/test/endtoend/utils/utils.go b/go/test/endtoend/utils/utils.go
index caec54d6aa..e1b729dd48 100644
--- a/go/test/endtoend/utils/utils.go
+++ b/go/test/endtoend/utils/utils.go
@@ -118,7 +118,7 @@ func ExecCompareMySQL(t *testing.T, vtConn, mysqlConn *mysql.Conn, query string)
 
 	mysqlQr, err := mysqlConn.ExecuteFetch(query, 1000, true)
 	require.NoError(t, err, "[MySQL Error] for query: "+query)
-	compareVitessAndMySQLResults(t, query, vtQr, mysqlQr)
+	compareVitessAndMySQLResults(t, query, vtQr, mysqlQr, false)
 	return vtQr
 }
 
@@ -159,3 +159,41 @@ func AssertMatchesWithTimeout(t *testing.T, conn *mysql.Conn, query, expected st
 
 	}
 }
+
+// WaitForAuthoritative waits for a table to become authoritative
+func WaitForAuthoritative(t *testing.T, cluster *cluster.LocalProcessCluster, ks, tbl string) error {
+	timeout := time.After(10 * time.Second)
+	for {
+		select {
+		case <-timeout:
+			return fmt.Errorf("schema tracking didn't mark table t2 as authoritative until timeout")
+		default:
+			time.Sleep(1 * time.Second)
+			res, err := cluster.VtgateProcess.ReadVSchema()
+			require.NoError(t, err, res)
+			t2Map := getTableT2Map(res, ks, tbl)
+			authoritative, fieldPresent := t2Map["column_list_authoritative"]
+			if !fieldPresent {
+				continue
+			}
+			authoritativeBool, isBool := authoritative.(bool)
+			if !isBool || !authoritativeBool {
+				continue
+			}
+			return nil
+		}
+	}
+}
+
+func getTableT2Map(res *interface{}, ks, tbl string) map[string]interface{} {
+	step1 := convertToMap(*res)["keyspaces"]
+	step2 := convertToMap(step1)[ks]
+	step3 := convertToMap(step2)["tables"]
+	tblMap := convertToMap(step3)[tbl]
+	return convertToMap(tblMap)
+}
+
+func convertToMap(input interface{}) map[string]interface{} {
+	output := input.(map[string]interface{})
+	return output
+}
diff --git a/go/test/endtoend/vtgate/gen4/gen4_test.go b/go/test/endtoend/vtgate/gen4/gen4_test.go
index 62a6a56a55..524958f579 100644
--- a/go/test/endtoend/vtgate/gen4/gen4_test.go
+++ b/go/test/endtoend/vtgate/gen4/gen4_test.go
@@ -32,51 +32,35 @@ import (
 )
 
 func TestOrderBy(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	defer func() {
-		_, _ = utils.ExecAllowError(t, conn, `delete from t1`)
-	}()
+	mcmp, closer := start(t)
+	defer closer()
 
 	// insert some data.
-	utils.Exec(t, conn, `insert into t1(id, col) values (100, 123),(10, 12),(1, 13),(1000, 1234)`)
+	utils.Exec(t, mcmp.VtConn, `insert into t1(id, col) values (100, 123),(10, 12),(1, 13),(1000, 1234)`)
 
 	// Gen4 only supported query.
-	utils.AssertMatches(t, conn, `select col from t1 order by id`, `[[INT64(13)] [INT64(12)] [INT64(123)] [INT64(1234)]]`)
+	utils.AssertMatches(t, mcmp.VtConn, `select col from t1 order by id`, `[[INT64(13)] [INT64(12)] [INT64(123)] [INT64(1234)]]`)
 
 	// Gen4 unsupported query. v3 supported.
-	utils.AssertMatches(t, conn, `select col from t1 order by 1`, `[[INT64(12)] [INT64(13)] [INT64(123)] [INT64(1234)]]`)
-
-	// unsupported in v3 and Gen4.
-	_, err = utils.ExecAllowError(t, conn, `select t1.* from t1 order by id`)
-	require.Error(t, err)
+	utils.AssertMatches(t, mcmp.VtConn, `select col from t1 order by 1`, `[[INT64(12)] [INT64(13)] [INT64(123)] [INT64(1234)]]`)
 }
 
 func TestCorrelatedExistsSubquery(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
+	mcmp, closer := start(t)
+	defer closer()
 
-	defer func() {
-		_, _ = utils.ExecAllowError(t, conn, `delete from t1`)
-		_, _ = utils.ExecAllowError(t, conn, `delete from t2`)
-	}()
 	// insert some data.
-	utils.Exec(t, conn, `insert into t1(id, col) values (100, 123), (10, 12), (1, 13), (4, 13), (1000, 1234)`)
-	utils.Exec(t, conn, `insert into t2(id, tcol1, tcol2) values (100, 13, 1),(9, 7, 15),(1, 123, 123),(1004, 134, 123)`)
+	utils.Exec(t, mcmp.VtConn, `insert into t1(id, col) values (100, 123), (10, 12), (1, 13), (4, 13), (1000, 1234)`)
+	utils.Exec(t, mcmp.VtConn, `insert into t2(id, tcol1, tcol2) values (100, 13, 1),(9, 7, 15),(1, 123, 123),(1004, 134, 123)`)
 
-	utils.AssertMatches(t, conn, `select id from t1 where exists(select 1 from t2 where t1.col = t2.tcol2)`,
+	utils.AssertMatches(t, mcmp.VtConn, `select id from t1 where exists(select 1 from t2 where t1.col = t2.tcol2)`,
 		`[[INT64(100)]]`)
-	utils.AssertMatches(t, conn, `select id from t1 where exists(select 1 from t2 where t1.col = t2.tcol1) order by id`,
+	utils.AssertMatches(t, mcmp.VtConn, `select id from t1 where exists(select 1 from t2 where t1.col = t2.tcol1) order by id`,
 		`[[INT64(1)] [INT64(4)] [INT64(100)]]`)
-	utils.AssertMatches(t, conn, `select id from t1 where id in (select id from t2) order by id`,
+	utils.AssertMatches(t, mcmp.VtConn, `select id from t1 where id in (select id from t2) order by id`,
 		`[[INT64(1)] [INT64(100)]]`)
 
-	utils.AssertMatches(t, conn, `
+	utils.AssertMatches(t, mcmp.VtConn, `
 select id 
 from t1 
 where exists(
@@ -86,7 +70,7 @@ where exists(
     having count(*) > 0
 )`,
 		`[[INT64(100)]]`)
-	utils.AssertMatches(t, conn, `
+	utils.AssertMatches(t, mcmp.VtConn, `
 select id 
 from t1 
 where exists(
@@ -95,7 +79,7 @@ where exists(
 	where t1.col = t2.tcol1
 ) order by id`,
 		`[[INT64(1)] [INT64(4)] [INT64(100)]]`)
-	utils.AssertMatchesNoOrder(t, conn, `
+	utils.AssertMatchesNoOrder(t, mcmp.VtConn, `
 select id 
 from t1 
 where exists(
@@ -107,31 +91,24 @@ where exists(
 }
 
 func TestGroupBy(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	defer func() {
-		_, _ = utils.ExecAllowError(t, conn, `delete from t1`)
-		_, _ = utils.ExecAllowError(t, conn, `delete from t2`)
-	}()
+	mcmp, closer := start(t)
+	defer closer()
 
 	// insert some data.
-	utils.Exec(t, conn, `insert into t1(id, col) values (1, 123),(2, 12),(3, 13),(4, 1234)`)
-	utils.Exec(t, conn, `insert into t2(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'B')`)
+	utils.Exec(t, mcmp.VtConn, `insert into t1(id, col) values (1, 123),(2, 12),(3, 13),(4, 1234)`)
+	utils.Exec(t, mcmp.VtConn, `insert into t2(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'B')`)
 
 	// Gen4 only supported query.
-	utils.AssertMatches(t, conn, `select tcol2, tcol1, count(id) from t2 group by tcol2, tcol1`,
+	utils.AssertMatches(t, mcmp.VtConn, `select tcol2, tcol1, count(id) from t2 group by tcol2, tcol1`,
 		`[[VARCHAR("A") VARCHAR("A") INT64(2)] [VARCHAR("A") VARCHAR("B") INT64(1)] [VARCHAR("A") VARCHAR("C") INT64(1)] [VARCHAR("B") VARCHAR("C") INT64(1)] [VARCHAR("C") VARCHAR("A") INT64(1)] [VARCHAR("C") VARCHAR("B") INT64(2)]]`)
 
-	utils.AssertMatches(t, conn, `select tcol1, tcol1 from t2 order by tcol1`,
+	utils.AssertMatches(t, mcmp.VtConn, `select tcol1, tcol1 from t2 order by tcol1`,
 		`[[VARCHAR("A") VARCHAR("A")] [VARCHAR("A") VARCHAR("A")] [VARCHAR("A") VARCHAR("A")] [VARCHAR("B") VARCHAR("B")] [VARCHAR("B") VARCHAR("B")] [VARCHAR("B") VARCHAR("B")] [VARCHAR("C") VARCHAR("C")] [VARCHAR("C") VARCHAR("C")]]`)
 
-	utils.AssertMatches(t, conn, `select tcol1, tcol1 from t1 join t2 on t1.id = t2.id order by tcol1`,
+	utils.AssertMatches(t, mcmp.VtConn, `select tcol1, tcol1 from t1 join t2 on t1.id = t2.id order by tcol1`,
 		`[[VARCHAR("A") VARCHAR("A")] [VARCHAR("A") VARCHAR("A")] [VARCHAR("B") VARCHAR("B")] [VARCHAR("C") VARCHAR("C")]]`)
 
-	utils.AssertMatches(t, conn, `select count(*) k, tcol1, tcol2, "abc" b from t2 group by tcol1, tcol2, b order by k, tcol2, tcol1`,
+	utils.AssertMatches(t, mcmp.VtConn, `select count(*) k, tcol1, tcol2, "abc" b from t2 group by tcol1, tcol2, b order by k, tcol2, tcol1`,
 		`[[INT64(1) VARCHAR("B") VARCHAR("A") VARCHAR("abc")] `+
 			`[INT64(1) VARCHAR("C") VARCHAR("A") VARCHAR("abc")] `+
 			`[INT64(1) VARCHAR("C") VARCHAR("B") VARCHAR("abc")] `+
@@ -141,180 +118,143 @@ func TestGroupBy(t *testing.T) {
 }
 
 func TestJoinBindVars(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	defer func() {
-		_, _ = utils.ExecAllowError(t, conn, `delete from t2`)
-		_, _ = utils.ExecAllowError(t, conn, `delete from t3`)
-	}()
+	mcmp, closer := start(t)
+	defer closer()
 
-	utils.Exec(t, conn, `insert into t2(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'B')`)
-	utils.Exec(t, conn, `insert into t3(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'B')`)
+	utils.Exec(t, mcmp.VtConn, `insert into t2(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'B')`)
+	utils.Exec(t, mcmp.VtConn, `insert into t3(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'B')`)
 
-	utils.AssertMatches(t, conn, `select t2.tcol1 from t2 join t3 on t2.tcol2 = t3.tcol2 where t2.tcol1 = 'A'`, `[[VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")]]`)
+	utils.AssertMatches(t, mcmp.VtConn, `select t2.tcol1 from t2 join t3 on t2.tcol2 = t3.tcol2 where t2.tcol1 = 'A'`, `[[VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")] [VARCHAR("A")]]`)
 }
 
 func TestDistinctAggregationFunc(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	defer utils.ExecAllowError(t, conn, `delete from t2`)
+	mcmp, closer := start(t)
+	defer closer()
 
 	// insert some data.
-	utils.Exec(t, conn, `insert into t2(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'A')`)
+	utils.Exec(t, mcmp.VtConn, `insert into t2(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'A')`)
 
 	// count on primary vindex
-	utils.AssertMatches(t, conn, `select tcol1, count(distinct id) from t2 group by tcol1`,
+	utils.AssertMatches(t, mcmp.VtConn, `select tcol1, count(distinct id) from t2 group by tcol1`,
 		`[[VARCHAR("A") INT64(3)] [VARCHAR("B") INT64(3)] [VARCHAR("C") INT64(2)]]`)
 
 	// count on any column
-	utils.AssertMatches(t, conn, `select tcol1, count(distinct tcol2) from t2 group by tcol1`,
+	utils.AssertMatches(t, mcmp.VtConn, `select tcol1, count(distinct tcol2) from t2 group by tcol1`,
 		`[[VARCHAR("A") INT64(2)] [VARCHAR("B") INT64(2)] [VARCHAR("C") INT64(1)]]`)
 
 	// sum of columns
-	utils.AssertMatches(t, conn, `select sum(id), sum(tcol1) from t2`,
+	utils.AssertMatches(t, mcmp.VtConn, `select sum(id), sum(tcol1) from t2`,
 		`[[DECIMAL(36) FLOAT64(0)]]`)
 
 	// sum on primary vindex
-	utils.AssertMatches(t, conn, `select tcol1, sum(distinct id) from t2 group by tcol1`,
+	utils.AssertMatches(t, mcmp.VtConn, `select tcol1, sum(distinct id) from t2 group by tcol1`,
 		`[[VARCHAR("A") DECIMAL(9)] [VARCHAR("B") DECIMAL(15)] [VARCHAR("C") DECIMAL(12)]]`)
 
 	// sum on any column
-	utils.AssertMatches(t, conn, `select tcol1, sum(distinct tcol2) from t2 group by tcol1`,
+	utils.AssertMatches(t, mcmp.VtConn, `select tcol1, sum(distinct tcol2) from t2 group by tcol1`,
 		`[[VARCHAR("A") DECIMAL(0)] [VARCHAR("B") DECIMAL(0)] [VARCHAR("C") DECIMAL(0)]]`)
 
 	// insert more data to get values on sum
-	utils.Exec(t, conn, `insert into t2(id, tcol1, tcol2) values (9, 'AA', null),(10, 'AA', '4'),(11, 'AA', '4'),(12, null, '5'),(13, null, '6'),(14, 'BB', '10'),(15, 'BB', '20'),(16, 'BB', 'X')`)
+	utils.Exec(t, mcmp.VtConn, `insert into t2(id, tcol1, tcol2) values (9, 'AA', null),(10, 'AA', '4'),(11, 'AA', '4'),(12, null, '5'),(13, null, '6'),(14, 'BB', '10'),(15, 'BB', '20'),(16, 'BB', 'X')`)
 
 	// multi distinct
-	utils.AssertMatches(t, conn, `select tcol1, count(distinct tcol2), sum(distinct tcol2) from t2 group by tcol1`,
+	utils.AssertMatches(t, mcmp.VtConn, `select tcol1, count(distinct tcol2), sum(distinct tcol2) from t2 group by tcol1`,
 		`[[NULL INT64(2) DECIMAL(11)] [VARCHAR("A") INT64(2) DECIMAL(0)] [VARCHAR("AA") INT64(1) DECIMAL(4)] [VARCHAR("B") INT64(2) DECIMAL(0)] [VARCHAR("BB") INT64(3) DECIMAL(30)] [VARCHAR("C") INT64(1) DECIMAL(0)]]`)
 }
 
 func TestDistinct(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	defer utils.ExecAllowError(t, conn, `delete from t2`)
+	mcmp, closer := start(t)
+	defer closer()
 
 	// insert some data.
-	utils.Exec(t, conn, `insert into t2(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'A')`)
+	utils.Exec(t, mcmp.VtConn, `insert into t2(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'A')`)
 
 	// multi distinct
-	utils.AssertMatches(t, conn, `select distinct tcol1, tcol2 from t2`,
+	utils.AssertMatches(t, mcmp.VtConn, `select distinct tcol1, tcol2 from t2`,
 		`[[VARCHAR("A") VARCHAR("A")] [VARCHAR("A") VARCHAR("C")] [VARCHAR("B") VARCHAR("A")] [VARCHAR("B") VARCHAR("C")] [VARCHAR("C") VARCHAR("A")]]`)
 }
 
 func TestSubQueries(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	defer func() {
-		_, _ = utils.ExecAllowError(t, conn, `delete from t2`)
-		_, _ = utils.ExecAllowError(t, conn, `delete from t3`)
-		_, _ = utils.ExecAllowError(t, conn, `delete from u_a`)
-	}()
+	mcmp, closer := start(t)
+	defer closer()
 
-	utils.Exec(t, conn, `insert into t2(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'B')`)
-	utils.Exec(t, conn, `insert into t3(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'B')`)
+	utils.Exec(t, mcmp.VtConn, `insert into t2(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'B')`)
+	utils.Exec(t, mcmp.VtConn, `insert into t3(id, tcol1, tcol2) values (1, 'A', 'A'),(2, 'B', 'C'),(3, 'A', 'C'),(4, 'C', 'A'),(5, 'A', 'A'),(6, 'B', 'C'),(7, 'B', 'A'),(8, 'C', 'B')`)
 
-	utils.AssertMatches(t, conn, `select t2.tcol1, t2.tcol2 from t2 where t2.id IN (select id from t3) order by t2.id`, `[[VARCHAR("A") VARCHAR("A")] [VARCHAR("B") VARCHAR("C")] [VARCHAR("A") VARCHAR("C")] [VARCHAR("C") VARCHAR("A")] [VARCHAR("A") VARCHAR("A")] [VARCHAR("B") VARCHAR("C")] [VARCHAR("B") VARCHAR("A")] [VARCHAR("C") VARCHAR("B")]]`)
-	utils.AssertMatches(t, conn, `select t2.tcol1, t2.tcol2 from t2 where t2.id IN (select t3.id from t3 join t2 on t2.id = t3.id) order by t2.id`, `[[VARCHAR("A") VARCHAR("A")] [VARCHAR("B") VARCHAR("C")] [VARCHAR("A") VARCHAR("C")] [VARCHAR("C") VARCHAR("A")] [VARCHAR("A") VARCHAR("A")] [VARCHAR("B") VARCHAR("C")] [VARCHAR("B") VARCHAR("A")] [VARCHAR("C") VARCHAR("B")]]`)
+	utils.AssertMatches(t, mcmp.VtConn, `select t2.tcol1, t2.tcol2 from t2 where t2.id IN (select id from t3) order by t2.id`, `[[VARCHAR("A") VARCHAR("A")] [VARCHAR("B") VARCHAR("C")] [VARCHAR("A") VARCHAR("C")] [VARCHAR("C") VARCHAR("A")] [VARCHAR("A") VARCHAR("A")] [VARCHAR("B") VARCHAR("C")] [VARCHAR("B") VARCHAR("A")] [VARCHAR("C") VARCHAR("B")]]`)
+	utils.AssertMatches(t, mcmp.VtConn, `select t2.tcol1, t2.tcol2 from t2 where t2.id IN (select t3.id from t3 join t2 on t2.id = t3.id) order by t2.id`, `[[VARCHAR("A") VARCHAR("A")] [VARCHAR("B") VARCHAR("C")] [VARCHAR("A") VARCHAR("C")] [VARCHAR("C") VARCHAR("A")] [VARCHAR("A") VARCHAR("A")] [VARCHAR("B") VARCHAR("C")] [VARCHAR("B") VARCHAR("A")] [VARCHAR("C") VARCHAR("B")]]`)
 
-	utils.AssertMatches(t, conn, `select u_a.a from u_a left join t2 on t2.id IN (select id from t2)`, `[]`)
+	utils.AssertMatches(t, mcmp.VtConn, `select u_a.a from u_a left join t2 on t2.id IN (select id from t2)`, `[]`)
 	// inserting some data in u_a
-	utils.Exec(t, conn, `insert into u_a(id, a) values (1, 1)`)
+	utils.Exec(t, mcmp.VtConn, `insert into u_a(id, a) values (1, 1)`)
 
 	// execute same query again.
-	qr := utils.Exec(t, conn, `select u_a.a from u_a left join t2 on t2.id IN (select id from t2)`)
+	qr := utils.Exec(t, mcmp.VtConn, `select u_a.a from u_a left join t2 on t2.id IN (select id from t2)`)
 	assert.EqualValues(t, 8, len(qr.Rows))
 	for index, row := range qr.Rows {
 		assert.EqualValues(t, `[INT64(1)]`, fmt.Sprintf("%v", row), "does not match for row: %d", index+1)
 	}
 
 	// fail as projection subquery is not scalar
-	_, err = utils.ExecAllowError(t, conn, `select (select id from t2) from t2 order by id`)
+	_, err := utils.ExecAllowError(t, mcmp.VtConn, `select (select id from t2) from t2 order by id`)
 	assert.EqualError(t, err, "subquery returned more than one row (errno 1105) (sqlstate HY000) during query: select (select id from t2) from t2 order by id")
 
-	utils.AssertMatches(t, conn, `select (select id from t2 order by id limit 1) from t2 order by id limit 2`, `[[INT64(1)] [INT64(1)]]`)
+	utils.AssertMatches(t, mcmp.VtConn, `select (select id from t2 order by id limit 1) from t2 order by id limit 2`, `[[INT64(1)] [INT64(1)]]`)
 }
 
 func TestPlannerWarning(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
+	mcmp, closer := start(t)
+	defer closer()
 
 	// straight_join query
-	_ = utils.Exec(t, conn, `select 1 from t1 straight_join t2 on t1.id = t2.id`)
-	utils.AssertMatches(t, conn, `show warnings`, `[[VARCHAR("Warning") UINT16(1235) VARCHAR("straight join is converted to normal join")]]`)
+	_ = utils.Exec(t, mcmp.VtConn, `select 1 from t1 straight_join t2 on t1.id = t2.id`)
+	utils.AssertMatches(t, mcmp.VtConn, `show warnings`, `[[VARCHAR("Warning") UINT16(1235) VARCHAR("straight join is converted to normal join")]]`)
 
 	// execute same query again.
-	_ = utils.Exec(t, conn, `select 1 from t1 straight_join t2 on t1.id = t2.id`)
-	utils.AssertMatches(t, conn, `show warnings`, `[[VARCHAR("Warning") UINT16(1235) VARCHAR("straight join is converted to normal join")]]`)
+	_ = utils.Exec(t, mcmp.VtConn, `select 1 from t1 straight_join t2 on t1.id = t2.id`)
+	utils.AssertMatches(t, mcmp.VtConn, `show warnings`, `[[VARCHAR("Warning") UINT16(1235) VARCHAR("straight join is converted to normal join")]]`)
 
 	// random query to reset the warning.
-	_ = utils.Exec(t, conn, `select 1 from t1`)
+	_ = utils.Exec(t, mcmp.VtConn, `select 1 from t1`)
 
 	// execute same query again.
-	_ = utils.Exec(t, conn, `select 1 from t1 straight_join t2 on t1.id = t2.id`)
-	utils.AssertMatches(t, conn, `show warnings`, `[[VARCHAR("Warning") UINT16(1235) VARCHAR("straight join is converted to normal join")]]`)
+	_ = utils.Exec(t, mcmp.VtConn, `select 1 from t1 straight_join t2 on t1.id = t2.id`)
+	utils.AssertMatches(t, mcmp.VtConn, `show warnings`, `[[VARCHAR("Warning") UINT16(1235) VARCHAR("straight join is converted to normal join")]]`)
 }
 
 func TestHashJoin(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
+	mcmp, closer := start(t)
+	defer closer()
 
-	defer func() {
-		_, _ = utils.ExecAllowError(t, conn, `delete from t1`)
-	}()
+	utils.Exec(t, mcmp.VtConn, `insert into t1(id, col) values (1, 1),(2, 3),(3, 4),(4, 7)`)
 
-	utils.Exec(t, conn, `insert into t1(id, col) values (1, 1),(2, 3),(3, 4),(4, 7)`)
+	utils.AssertMatches(t, mcmp.VtConn, `select /*vt+ ALLOW_HASH_JOIN */ t1.id from t1 x join t1 where x.col = t1.col and x.id <= 3 and t1.id >= 3`, `[[INT64(3)]]`)
 
-	utils.AssertMatches(t, conn, `select /*vt+ ALLOW_HASH_JOIN */ t1.id from t1 x join t1 where x.col = t1.col and x.id <= 3 and t1.id >= 3`, `[[INT64(3)]]`)
-
-	utils.Exec(t, conn, `set workload = olap`)
-	defer utils.Exec(t, conn, `set workload = oltp`)
-	utils.AssertMatches(t, conn, `select /*vt+ ALLOW_HASH_JOIN */ t1.id from t1 x join t1 where x.col = t1.col and x.id <= 3 and t1.id >= 3`, `[[INT64(3)]]`)
+	utils.Exec(t, mcmp.VtConn, `set workload = olap`)
+	defer utils.Exec(t, mcmp.VtConn, `set workload = oltp`)
+	utils.AssertMatches(t, mcmp.VtConn, `select /*vt+ ALLOW_HASH_JOIN */ t1.id from t1 x join t1 where x.col = t1.col and x.id <= 3 and t1.id >= 3`, `[[INT64(3)]]`)
 }
 
 func TestMultiColumnVindex(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	defer utils.ExecAllowError(t, conn, `delete from user_region`)
-	utils.Exec(t, conn, `insert into user_region(id, cola, colb) values (1, 1, 2),(2, 30, 40),(3, 500, 600),(4, 30, 40),(5, 10000, 30000),(6, 422333, 40),(7, 30, 60)`)
+	mcmp, closer := start(t)
+	defer closer()
+	mcmp.Exec(`insert into user_region(id, cola, colb) values (1, 1, 2),(2, 30, 40),(3, 500, 600),(4, 30, 40),(5, 10000, 30000),(6, 422333, 40),(7, 30, 60)`)
 
 	for _, workload := range []string{"olap", "oltp"} {
 		t.Run(workload, func(t *testing.T) {
-			utils.Exec(t, conn, fmt.Sprintf(`set workload = %s`, workload))
-			utils.AssertMatches(t, conn, `select id from user_region where cola = 1 and colb = 2`, `[[INT64(1)]]`)
-			utils.AssertMatches(t, conn, `select id from user_region where cola in (30,422333) and colb = 40 order by id`, `[[INT64(2)] [INT64(4)] [INT64(6)]]`)
-			utils.AssertMatches(t, conn, `select id from user_region where cola in (30,422333) and colb in (40,60) order by id`, `[[INT64(2)] [INT64(4)] [INT64(6)] [INT64(7)]]`)
-			utils.AssertMatches(t, conn, `select id from user_region where cola in (30,422333) and colb in (40,60) and cola = 422333`, `[[INT64(6)]]`)
-			utils.AssertMatches(t, conn, `select id from user_region where cola in (30,422333) and colb in (40,60) and cola = 30 and colb = 60`, `[[INT64(7)]]`)
+			utils.Exec(t, mcmp.VtConn, fmt.Sprintf(`set workload = %s`, workload))
+			utils.AssertMatches(t, mcmp.VtConn, `select id from user_region where cola = 1 and colb = 2`, `[[INT64(1)]]`)
+			utils.AssertMatches(t, mcmp.VtConn, `select id from user_region where cola in (30,422333) and colb = 40 order by id`, `[[INT64(2)] [INT64(4)] [INT64(6)]]`)
+			utils.AssertMatches(t, mcmp.VtConn, `select id from user_region where cola in (30,422333) and colb in (40,60) order by id`, `[[INT64(2)] [INT64(4)] [INT64(6)] [INT64(7)]]`)
+			utils.AssertMatches(t, mcmp.VtConn, `select id from user_region where cola in (30,422333) and colb in (40,60) and cola = 422333`, `[[INT64(6)]]`)
+			utils.AssertMatches(t, mcmp.VtConn, `select id from user_region where cola in (30,422333) and colb in (40,60) and cola = 30 and colb = 60`, `[[INT64(7)]]`)
 		})
 	}
 }
 
 func TestFanoutVindex(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
+	mcmp, closer := start(t)
+	defer closer()
 
 	tcases := []struct {
 		regionID int
@@ -336,18 +276,18 @@ func TestFanoutVindex(t *testing.T) {
 		exp:      `[[INT64(33) INT64(20) VARCHAR("shard-20c0-")]]`,
 	}}
 
-	defer utils.ExecAllowError(t, conn, `delete from region_tbl`)
+	defer utils.ExecAllowError(t, mcmp.VtConn, `delete from region_tbl`)
 	uid := 1
 	// insert data in all shards to know where the query fan-out
 	for _, s := range shardedKsShards {
-		utils.Exec(t, conn, fmt.Sprintf("use `%s:%s`", shardedKs, s))
+		utils.Exec(t, mcmp.VtConn, fmt.Sprintf("use `%s:%s`", shardedKs, s))
 		for _, tcase := range tcases {
-			utils.Exec(t, conn, fmt.Sprintf("insert into region_tbl(rg,uid,msg) values(%d,%d,'shard-%s')", tcase.regionID, uid, s))
+			utils.Exec(t, mcmp.VtConn, fmt.Sprintf("insert into region_tbl(rg,uid,msg) values(%d,%d,'shard-%s')", tcase.regionID, uid, s))
 			uid++
 		}
 	}
 
-	newConn, err := mysql.Connect(ctx, &vtParams)
+	newConn, err := mysql.Connect(context.Background(), &vtParams)
 	require.NoError(t, err)
 	defer newConn.Close()
 
@@ -363,10 +303,8 @@ func TestFanoutVindex(t *testing.T) {
 }
 
 func TestSubShardVindex(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
+	mcmp, closer := start(t)
+	defer closer()
 
 	tcases := []struct {
 		regionID int
@@ -397,14 +335,14 @@ func TestSubShardVindex(t *testing.T) {
 	uid := 1
 	// insert data in all shards to know where the query fan-out
 	for _, s := range shardedKsShards {
-		utils.Exec(t, conn, fmt.Sprintf("use `%s:%s`", shardedKs, s))
+		utils.Exec(t, mcmp.VtConn, fmt.Sprintf("use `%s:%s`", shardedKs, s))
 		for _, tcase := range tcases {
-			utils.Exec(t, conn, fmt.Sprintf("insert into multicol_tbl(cola,colb,colc,msg) values(%d,_binary '%d','%d','shard-%s')", tcase.regionID, uid, uid, s))
+			utils.Exec(t, mcmp.VtConn, fmt.Sprintf("insert into multicol_tbl(cola,colb,colc,msg) values(%d,_binary '%d','%d','shard-%s')", tcase.regionID, uid, uid, s))
 			uid++
 		}
 	}
 
-	newConn, err := mysql.Connect(ctx, &vtParams)
+	newConn, err := mysql.Connect(context.Background(), &vtParams)
 	require.NoError(t, err)
 	defer newConn.Close()
 
@@ -421,10 +359,8 @@ func TestSubShardVindex(t *testing.T) {
 }
 
 func TestSubShardVindexDML(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
+	mcmp, closer := start(t)
+	defer closer()
 
 	tcases := []struct {
 		regionID       int
@@ -449,14 +385,14 @@ func TestSubShardVindexDML(t *testing.T) {
 	uid := 1
 	// insert data in all shards to know where the query fan-out
 	for _, s := range shardedKsShards {
-		utils.Exec(t, conn, fmt.Sprintf("use `%s:%s`", shardedKs, s))
+		utils.Exec(t, mcmp.VtConn, fmt.Sprintf("use `%s:%s`", shardedKs, s))
 		for _, tcase := range tcases {
-			utils.Exec(t, conn, fmt.Sprintf("insert into multicol_tbl(cola,colb,colc,msg) values(%d,_binary '%d','%d','shard-%s')", tcase.regionID, uid, uid, s))
+			utils.Exec(t, mcmp.VtConn, fmt.Sprintf("insert into multicol_tbl(cola,colb,colc,msg) values(%d,_binary '%d','%d','shard-%s')", tcase.regionID, uid, uid, s))
 			uid++
 		}
 	}
 
-	newConn, err := mysql.Connect(ctx, &vtParams)
+	newConn, err := mysql.Connect(context.Background(), &vtParams)
 	require.NoError(t, err)
 	defer newConn.Close()
 
@@ -477,27 +413,44 @@ func TestSubShardVindexDML(t *testing.T) {
 }
 
 func TestOuterJoin(t *testing.T) {
-	ctx := context.Background()
-	conn, err := mysql.Connect(ctx, &vtParams)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	_, _ = utils.ExecAllowError(t, conn, `delete from t1`)
-	_, _ = utils.ExecAllowError(t, conn, `delete from t2`)
-	defer func() {
-		_, _ = utils.ExecAllowError(t, conn, `delete from t1`)
-		_, _ = utils.ExecAllowError(t, conn, `delete from t2`)
-	}()
+	mcmp, closer := start(t)
+	defer closer()
 
 	// insert some data.
-	utils.Exec(t, conn, `insert into t1(id, col) values (100, 123), (10, 123), (1, 13), (1000, 1234)`)
-	utils.Exec(t, conn, `insert into t2(id, tcol1, tcol2) values (12, 13, 1),(123, 7, 15),(1, 123, 123),(1004, 134, 123)`)
+	utils.Exec(t, mcmp.VtConn, `insert into t1(id, col) values (100, 123), (10, 123), (1, 13), (1000, 1234)`)
+	utils.Exec(t, mcmp.VtConn, `insert into t2(id, tcol1, tcol2) values (12, 13, 1),(123, 7, 15),(1, 123, 123),(1004, 134, 123)`)
 
 	// Gen4 only supported query.
-	utils.AssertMatchesNoOrder(t, conn, `select t1.id, t2.tcol1+t2.tcol2 from t1 left join t2 on t1.col = t2.id`, `[[INT64(10) FLOAT64(22)] [INT64(1) NULL] [INT64(100) FLOAT64(22)] [INT64(1000) NULL]]`)
-	utils.AssertMatchesNoOrder(t, conn, `select t1.id, t2.id, t2.tcol1+t1.col+t2.tcol2 from t1 left join t2 on t1.col = t2.id`,
+	utils.AssertMatchesNoOrder(t, mcmp.VtConn, `select t1.id, t2.tcol1+t2.tcol2 from t1 left join t2 on t1.col = t2.id`, `[[INT64(10) FLOAT64(22)] [INT64(1) NULL] [INT64(100) FLOAT64(22)] [INT64(1000) NULL]]`)
+	utils.AssertMatchesNoOrder(t, mcmp.VtConn, `select t1.id, t2.id, t2.tcol1+t1.col+t2.tcol2 from t1 left join t2 on t1.col = t2.id`,
 		`[[INT64(10) INT64(123) FLOAT64(145)]`+
 			` [INT64(1) NULL NULL]`+
 			` [INT64(100) INT64(123) FLOAT64(145)]`+
 			` [INT64(1000) NULL NULL]]`)
 }
+
+func TestUsingJoin(t *testing.T) {
+	require.NoError(t, utils.WaitForAuthoritative(t, clusterInstance, shardedKs, "t1"))
+	require.NoError(t, utils.WaitForAuthoritative(t, clusterInstance, shardedKs, "t2"))
+	require.NoError(t, utils.WaitForAuthoritative(t, clusterInstance, shardedKs, "t3"))
+
+	mcmp, closer := start(t)
+	defer closer()
+
+	// insert some data.
+	mcmp.Exec(`insert into t1(id, col) values          (1, 1), (2, 2), (3, 3), (5, 5)`)
+	mcmp.Exec(`insert into t2(id, tcol1, tcol2) values (1, 12, 12),(3, 3, 13),(4, 123, 123),(5, 134, 123)`)
+	mcmp.Exec(`insert into t3(id, tcol1, tcol2) values (1, 12, 12),(4, 123, 123),(5, 134, 123)`)
+
+	// Gen4 only supported query.
+	mcmp.AssertMatchesNoOrderInclColumnNames(`select t1.id from t1 join t2 using(id)`,
+		`[[INT64(1)] [INT64(3)] [INT64(5)]]`)
+	mcmp.AssertMatchesNoOrderInclColumnNames(`select t2.id from t2 join t3 using (id, tcol1, tcol2)`,
+		`[[INT64(1)] [INT64(4)] [INT64(5)]]`)
+	mcmp.AssertMatchesNoOrderInclColumnNames(`select * from t2 join t3 using (tcol1)`,
+		`[[VARCHAR("12") INT64(1) VARCHAR("12") INT64(1) VARCHAR("12")] `+
+			`[VARCHAR("123") INT64(4) VARCHAR("123") INT64(4) VARCHAR("123")] `+
+			`[VARCHAR("134") INT64(5) VARCHAR("123") INT64(5) VARCHAR("123")]]`)
+	mcmp.AssertMatchesNoOrderInclColumnNames(`select * from t2 join t3 using (tcol1) having tcol1 = 12`,
+		`[[VARCHAR("12") INT64(1) VARCHAR("12") INT64(1) VARCHAR("12")]]`)
+}
diff --git a/go/test/endtoend/vtgate/gen4/main_test.go b/go/test/endtoend/vtgate/gen4/main_test.go
index 79a0d4859f..8a2e1b22c5 100644
--- a/go/test/endtoend/vtgate/gen4/main_test.go
+++ b/go/test/endtoend/vtgate/gen4/main_test.go
@@ -18,9 +18,14 @@ package vtgate
 
 import (
 	"flag"
+	"fmt"
 	"os"
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/test/endtoend/utils"
+
 	"vitess.io/vitess/go/vt/vtgate/planbuilder"
 
 	"vitess.io/vitess/go/mysql"
@@ -30,6 +35,7 @@ import (
 var (
 	clusterInstance  *cluster.LocalProcessCluster
 	vtParams         mysql.ConnParams
+	mysqlParams      mysql.ConnParams
 	shardedKs        = "ks"
 	unshardedKs      = "uks"
 	shardedKsShards  = []string{"-19a0", "19a0-20", "20-20c0", "20c0-"}
@@ -214,6 +220,9 @@ func TestMain(m *testing.M) {
 			SchemaSQL: shardedSchemaSQL,
 			VSchema:   shardedVSchema,
 		}
+
+		clusterInstance.VtGateExtraArgs = []string{"--schema_change_signal"}
+		clusterInstance.VtTabletExtraArgs = []string{"--queryserver-config-schema-change-signal", "--queryserver-config-schema-change-signal-interval", "0.1"}
 		err = clusterInstance.StartKeyspace(*sKs, shardedKsShards, 0, false)
 		if err != nil {
 			return 1
@@ -250,7 +259,36 @@ func TestMain(m *testing.M) {
 			Host: clusterInstance.Hostname,
 			Port: clusterInstance.VtgateMySQLPort,
 		}
+
+		conn, closer, err := utils.NewMySQL(clusterInstance, shardedKs, shardedSchemaSQL)
+		if err != nil {
+			fmt.Println(err)
+			return 1
+		}
+		defer closer()
+		mysqlParams = conn
 		return m.Run()
 	}()
 	os.Exit(exitCode)
 }
+
+func start(t *testing.T) (utils.MySQLCompare, func()) {
+	mcmp, err := utils.NewMySQLCompare(t, vtParams, mysqlParams)
+	require.NoError(t, err)
+	deleteAll := func() {
+		_, _ = utils.ExecAllowError(t, mcmp.VtConn, "set workload = oltp")
+
+		tables := []string{"t1", "t2", "t3", "user_region", "region_tbl", "multicol_tbl", "t1_id2_idx", "t2_id4_idx", "u_a", "u_b"}
+		for _, table := range tables {
+			_, _ = mcmp.ExecAndIgnore("delete from " + table)
+		}
+	}
+
+	deleteAll()
+
+	return mcmp, func() {
+		deleteAll()
+		mcmp.Close()
+		cluster.PanicHandler(t)
+	}
+}
diff --git a/go/test/endtoend/vtgate/queries/foundrows/found_rows_test.go b/go/test/endtoend/vtgate/queries/foundrows/found_rows_test.go
index dd0056864d..314b83d502 100644
--- a/go/test/endtoend/vtgate/queries/foundrows/found_rows_test.go
+++ b/go/test/endtoend/vtgate/queries/foundrows/found_rows_test.go
@@ -17,9 +17,7 @@ limitations under the License.
 package foundrows
 
 import (
-	"fmt"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -46,7 +44,7 @@ func TestFoundRows(t *testing.T) {
 	// Wait for schema tracking to run and mark t2 as authoritative before we try out the queries.
 	// Some of the queries depend on schema tracking to run successfully to be able to replace the StarExpr
 	// in the select clause with the definitive column list.
-	err = waitForT2Authoritative(t)
+	err = utils.WaitForAuthoritative(t, clusterInstance, keyspaceName, "t2")
 	require.NoError(t, err)
 	runTests := func(workload string) {
 		mcmp.AssertFoundRowsValue("select * from t2", workload, 5)
@@ -61,38 +59,3 @@ func TestFoundRows(t *testing.T) {
 	utils.Exec(t, mcmp.VtConn, "set workload = olap")
 	runTests("olap")
 }
-
-// waitForT2Authoritative waits for the table t2 to become authoritative
-func waitForT2Authoritative(t *testing.T) error {
-	timeout := time.After(10 * time.Second)
-	for {
-		select {
-		case <-timeout:
-			return fmt.Errorf("schema tracking didn't mark table t2 as authoritative until timeout")
-		default:
-			time.Sleep(1 * time.Second)
-			res, err := clusterInstance.VtgateProcess.ReadVSchema()
-			require.NoError(t, err, res)
-			t2Map := getTableT2Map(res)
-			authoritative, fieldPresent := t2Map["column_list_authoritative"]
-			if !fieldPresent {
-				continue
-			}
-			authoritativeBool, isBool := authoritative.(bool)
-			if !isBool || !authoritativeBool {
-				continue
-			}
-			return nil
-		}
-	}
-}
-
-func getTableT2Map(res *interface{}) map[string]interface{} {
-	t2Map := convertToMap(convertToMap(convertToMap(convertToMap(convertToMap(*res)["keyspaces"])[keyspaceName])["tables"])["t2"])
-	return t2Map
-}
-
-func convertToMap(input interface{}) map[string]interface{} {
-	output := input.(map[string]interface{})
-	return output
-}
diff --git a/go/vt/schemadiff/table_test.go b/go/vt/schemadiff/table_test.go
index 25fad3648a..87f8134470 100644
--- a/go/vt/schemadiff/table_test.go
+++ b/go/vt/schemadiff/table_test.go
@@ -411,29 +411,29 @@ func TestCreateTableDiff(t *testing.T) {
 			name:  "change partitioning hash",
 			from:  "create table t1 (id int primary key) partition by hash (id) partitions 4",
 			to:    "create table t1 (id int primary key) partition by hash (id) partitions 5",
-			diff:  "alter table t1 partition by hash (id) partitions 5",
-			cdiff: "ALTER TABLE `t1` PARTITION BY HASH (`id`) PARTITIONS 5",
+			diff:  "alter table t1 \npartition by hash (id) partitions 5",
+			cdiff: "ALTER TABLE `t1` \nPARTITION BY HASH (`id`) PARTITIONS 5",
 		},
 		{
 			name:  "change partitioning key",
 			from:  "create table t1 (id int primary key) partition by key (id) partitions 2",
 			to:    "create table t1 (id int primary key) partition by hash (id) partitions 5",
-			diff:  "alter table t1 partition by hash (id) partitions 5",
-			cdiff: "ALTER TABLE `t1` PARTITION BY HASH (`id`) PARTITIONS 5",
+			diff:  "alter table t1 \npartition by hash (id) partitions 5",
+			cdiff: "ALTER TABLE `t1` \nPARTITION BY HASH (`id`) PARTITIONS 5",
 		},
 		{
 			name:  "change partitioning list",
 			from:  "create table t1 (id int primary key) partition by key (id) partitions 2",
 			to:    "create table t1 (id int primary key) partition by list (id) (partition p1 values in(11,21), partition p2 values in (12,22))",
-			diff:  "alter table t1 partition by list (id) (partition p1 values in (11, 21), partition p2 values in (12, 22))",
-			cdiff: "ALTER TABLE `t1` PARTITION BY LIST (`id`) (PARTITION `p1` VALUES IN (11, 21), PARTITION `p2` VALUES IN (12, 22))",
+			diff:  "alter table t1 \npartition by list (id)\n(partition p1 values in (11, 21),\n partition p2 values in (12, 22))",
+			cdiff: "ALTER TABLE `t1` \nPARTITION BY LIST (`id`)\n(PARTITION `p1` VALUES IN (11, 21),\n PARTITION `p2` VALUES IN (12, 22))",
 		},
 		{
 			name:  "change partitioning range: rotate",
 			from:  "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:    "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (20), partition p3 values less than (30), partition p4 values less than (40))",
-			diff:  "alter table t1 partition by range (id) (partition p2 values less than (20), partition p3 values less than (30), partition p4 values less than (40))",
-			cdiff: "ALTER TABLE `t1` PARTITION BY RANGE (`id`) (PARTITION `p2` VALUES LESS THAN (20), PARTITION `p3` VALUES LESS THAN (30), PARTITION `p4` VALUES LESS THAN (40))",
+			diff:  "alter table t1 \npartition by range (id)\n(partition p2 values less than (20),\n partition p3 values less than (30),\n partition p4 values less than (40))",
+			cdiff: "ALTER TABLE `t1` \nPARTITION BY RANGE (`id`)\n(PARTITION `p2` VALUES LESS THAN (20),\n PARTITION `p3` VALUES LESS THAN (30),\n PARTITION `p4` VALUES LESS THAN (40))",
 		},
 		{
 			name:     "change partitioning range: ignore rotate",
@@ -502,56 +502,56 @@ func TestCreateTableDiff(t *testing.T) {
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key, i int) partition by range (id) (partition p3 values less than (30))",
 			rotation: RangeRotationFullSpec,
-			diff:     "alter table t1 add column i int partition by range (id) (partition p3 values less than (30))",
-			cdiff:    "ALTER TABLE `t1` ADD COLUMN `i` int PARTITION BY RANGE (`id`) (PARTITION `p3` VALUES LESS THAN (30))",
+			diff:     "alter table t1 add column i int \npartition by range (id)\n(partition p3 values less than (30))",
+			cdiff:    "ALTER TABLE `t1` ADD COLUMN `i` int \nPARTITION BY RANGE (`id`)\n(PARTITION `p3` VALUES LESS THAN (30))",
 		},
 		{
 			name:     "change partitioning range: ignore rotate, not a rotation",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (25), partition p3 values less than (30), partition p4 values less than (40))",
 			rotation: RangeRotationIgnore,
-			diff:     "alter table t1 partition by range (id) (partition p2 values less than (25), partition p3 values less than (30), partition p4 values less than (40))",
-			cdiff:    "ALTER TABLE `t1` PARTITION BY RANGE (`id`) (PARTITION `p2` VALUES LESS THAN (25), PARTITION `p3` VALUES LESS THAN (30), PARTITION `p4` VALUES LESS THAN (40))",
+			diff:     "alter table t1 \npartition by range (id)\n(partition p2 values less than (25),\n partition p3 values less than (30),\n partition p4 values less than (40))",
+			cdiff:    "ALTER TABLE `t1` \nPARTITION BY RANGE (`id`)\n(PARTITION `p2` VALUES LESS THAN (25),\n PARTITION `p3` VALUES LESS THAN (30),\n PARTITION `p4` VALUES LESS THAN (40))",
 		},
 		{
 			name:     "change partitioning range: ignore rotate, not a rotation 2",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (20), partition p3 values less than (35), partition p4 values less than (40))",
 			rotation: RangeRotationIgnore,
-			diff:     "alter table t1 partition by range (id) (partition p2 values less than (20), partition p3 values less than (35), partition p4 values less than (40))",
-			cdiff:    "ALTER TABLE `t1` PARTITION BY RANGE (`id`) (PARTITION `p2` VALUES LESS THAN (20), PARTITION `p3` VALUES LESS THAN (35), PARTITION `p4` VALUES LESS THAN (40))",
+			diff:     "alter table t1 \npartition by range (id)\n(partition p2 values less than (20),\n partition p3 values less than (35),\n partition p4 values less than (40))",
+			cdiff:    "ALTER TABLE `t1` \nPARTITION BY RANGE (`id`)\n(PARTITION `p2` VALUES LESS THAN (20),\n PARTITION `p3` VALUES LESS THAN (35),\n PARTITION `p4` VALUES LESS THAN (40))",
 		},
 		{
 			name:     "change partitioning range: ignore rotate, not a rotation 3",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key) partition by range (id) (partition p2 values less than (20), partition pX values less than (30), partition p4 values less than (40))",
 			rotation: RangeRotationIgnore,
-			diff:     "alter table t1 partition by range (id) (partition p2 values less than (20), partition pX values less than (30), partition p4 values less than (40))",
-			cdiff:    "ALTER TABLE `t1` PARTITION BY RANGE (`id`) (PARTITION `p2` VALUES LESS THAN (20), PARTITION `pX` VALUES LESS THAN (30), PARTITION `p4` VALUES LESS THAN (40))",
+			diff:     "alter table t1 \npartition by range (id)\n(partition p2 values less than (20),\n partition pX values less than (30),\n partition p4 values less than (40))",
+			cdiff:    "ALTER TABLE `t1` \nPARTITION BY RANGE (`id`)\n(PARTITION `p2` VALUES LESS THAN (20),\n PARTITION `pX` VALUES LESS THAN (30),\n PARTITION `p4` VALUES LESS THAN (40))",
 		},
 		{
 			name:     "change partitioning range: ignore rotate, not a rotation 4",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key) partition by range (id) (partition pX values less than (20), partition p3 values less than (30), partition p4 values less than (40))",
 			rotation: RangeRotationIgnore,
-			diff:     "alter table t1 partition by range (id) (partition pX values less than (20), partition p3 values less than (30), partition p4 values less than (40))",
-			cdiff:    "ALTER TABLE `t1` PARTITION BY RANGE (`id`) (PARTITION `pX` VALUES LESS THAN (20), PARTITION `p3` VALUES LESS THAN (30), PARTITION `p4` VALUES LESS THAN (40))",
+			diff:     "alter table t1 \npartition by range (id)\n(partition pX values less than (20),\n partition p3 values less than (30),\n partition p4 values less than (40))",
+			cdiff:    "ALTER TABLE `t1` \nPARTITION BY RANGE (`id`)\n(PARTITION `pX` VALUES LESS THAN (20),\n PARTITION `p3` VALUES LESS THAN (30),\n PARTITION `p4` VALUES LESS THAN (40))",
 		},
 		{
 			name:     "change partitioning range: ignore rotate, nothing shared",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key) partition by range (id) (partition p4 values less than (40), partition p5 values less than (50), partition p6 values less than (60))",
 			rotation: RangeRotationIgnore,
-			diff:     "alter table t1 partition by range (id) (partition p4 values less than (40), partition p5 values less than (50), partition p6 values less than (60))",
-			cdiff:    "ALTER TABLE `t1` PARTITION BY RANGE (`id`) (PARTITION `p4` VALUES LESS THAN (40), PARTITION `p5` VALUES LESS THAN (50), PARTITION `p6` VALUES LESS THAN (60))",
+			diff:     "alter table t1 \npartition by range (id)\n(partition p4 values less than (40),\n partition p5 values less than (50),\n partition p6 values less than (60))",
+			cdiff:    "ALTER TABLE `t1` \nPARTITION BY RANGE (`id`)\n(PARTITION `p4` VALUES LESS THAN (40),\n PARTITION `p5` VALUES LESS THAN (50),\n PARTITION `p6` VALUES LESS THAN (60))",
 		},
 		{
 			name:     "change partitioning range: ignore rotate, no names shared, definitions shared",
 			from:     "create table t1 (id int primary key) partition by range (id) (partition p1 values less than (10), partition p2 values less than (20), partition p3 values less than (30))",
 			to:       "create table t1 (id int primary key) partition by range (id) (partition pA values less than (20), partition pB values less than (30), partition pC values less than (40))",
 			rotation: RangeRotationIgnore,
-			diff:     "alter table t1 partition by range (id) (partition pA values less than (20), partition pB values less than (30), partition pC values less than (40))",
-			cdiff:    "ALTER TABLE `t1` PARTITION BY RANGE (`id`) (PARTITION `pA` VALUES LESS THAN (20), PARTITION `pB` VALUES LESS THAN (30), PARTITION `pC` VALUES LESS THAN (40))",
+			diff:     "alter table t1 \npartition by range (id)\n(partition pA values less than (20),\n partition pB values less than (30),\n partition pC values less than (40))",
+			cdiff:    "ALTER TABLE `t1` \nPARTITION BY RANGE (`id`)\n(PARTITION `pA` VALUES LESS THAN (20),\n PARTITION `pB` VALUES LESS THAN (30),\n PARTITION `pC` VALUES LESS THAN (40))",
 		},
 
 		//
@@ -1217,7 +1217,7 @@ func TestNormalize(t *testing.T) {
 		{
 			name: "correct case for engine in partitions",
 			from: "create table a (id int not null primary key) engine InnoDB, charset utf8mb4, collate utf8mb4_0900_ai_ci partition by range (`id`) (partition `p10` values less than(10) engine innodb)",
-			to:   "CREATE TABLE `a` (\n\t`id` int NOT NULL PRIMARY KEY\n) ENGINE InnoDB,\n  CHARSET utf8mb4,\n  COLLATE utf8mb4_0900_ai_ci PARTITION BY RANGE (`id`) (PARTITION `p10` VALUES LESS THAN (10) ENGINE InnoDB)",
+			to:   "CREATE TABLE `a` (\n\t`id` int NOT NULL PRIMARY KEY\n) ENGINE InnoDB,\n  CHARSET utf8mb4,\n  COLLATE utf8mb4_0900_ai_ci\nPARTITION BY RANGE (`id`)\n(PARTITION `p10` VALUES LESS THAN (10) ENGINE InnoDB)",
 		},
 		{
 			name: "generates a name for checks",
diff --git a/go/vt/sqlparser/parse_test.go b/go/vt/sqlparser/parse_test.go
index f12cea9f87..a95dcb140c 100644
--- a/go/vt/sqlparser/parse_test.go
+++ b/go/vt/sqlparser/parse_test.go
@@ -1241,9 +1241,9 @@ var (
 		input:  "alter table t2 add primary key `zzz` (id)",
 		output: "alter table t2 add primary key (id)",
 	}, {
-		input: "alter table a partition by hash (id) partitions 4",
+		input: "alter table a \npartition by hash (id) partitions 4",
 	}, {
-		input: "alter table a partition by range (id) (partition p0 values less than (10), partition p1 values less than maxvalue)",
+		input: "alter table a \npartition by range (id)\n(partition p0 values less than (10),\n partition p1 values less than maxvalue)",
 	}, {
 		input:      "create database a garbage values",
 		output:     "create database a",
@@ -1434,28 +1434,28 @@ var (
 		input: "create table invalid_enum_value_name (\n\there_be_enum enum('$§!') default null\n)",
 	}, {
 		input:  "create table t (id int) partition by hash (id) partitions 3",
-		output: "create table t (\n\tid int\n) partition by hash (id) partitions 3",
+		output: "create table t (\n\tid int\n)\npartition by hash (id) partitions 3",
 	}, {
 		input:  "create table t (hired date) partition by linear hash (year(hired)) partitions 4",
-		output: "create table t (\n\thired date\n) partition by linear hash (year(hired)) partitions 4",
+		output: "create table t (\n\thired date\n)\npartition by linear hash (year(hired)) partitions 4",
 	}, {
 		input:  "create table t (id int) partition by key (id) partitions 2",
-		output: "create table t (\n\tid int\n) partition by key (id) partitions 2",
+		output: "create table t (\n\tid int\n)\npartition by key (id) partitions 2",
 	}, {
 		input:  "create table t (id int) partition by key algorithm = 1 (id)",
-		output: "create table t (\n\tid int\n) partition by key algorithm = 1 (id)",
+		output: "create table t (\n\tid int\n)\npartition by key algorithm = 1 (id)",
 	}, {
 		input:  "create table t (id int not null) partition by linear key (id) partitions 5",
-		output: "create table t (\n\tid int not null\n) partition by linear key (id) partitions 5",
+		output: "create table t (\n\tid int not null\n)\npartition by linear key (id) partitions 5",
 	}, {
-		input:  "create table t (id int) partition by list (id)",
-		output: "create table t (\n\tid int\n) partition by list (id)", // TODO PARTITION BY LIST(id) (PARTITION p0 VALUES IN (1, 4, 7))
+		input:  "create table t (id int) partition by list (id) (partition p0 values in (1, 4, 7))",
+		output: "create table t (\n\tid int\n)\npartition by list (id)\n(partition p0 values in (1, 4, 7))",
 	}, {
 		input:  "create table t (renewal date) partition by range columns (renewal) (partition p0 values less than ('2021-08-27'))",
-		output: "create table t (\n\trenewal date\n) partition by range columns (renewal) (partition p0 values less than ('2021-08-27'))",
+		output: "create table t (\n\trenewal date\n)\npartition by range columns (renewal)\n(partition p0 values less than ('2021-08-27'))",
 	}, {
 		input:  "create table t (pur date) partition by range (year(pur)) subpartition by hash (to_days(pur)) subpartitions 2 (partition p0 values less than (2015), partition p2 values less than (2018))",
-		output: "create table t (\n\tpur date\n) partition by range (year(pur)) subpartition by hash (to_days(pur)) subpartitions 2 (partition p0 values less than (2015), partition p2 values less than (2018))",
+		output: "create table t (\n\tpur date\n)\npartition by range (year(pur)) subpartition by hash (to_days(pur)) subpartitions 2\n(partition p0 values less than (2015),\n partition p2 values less than (2018))",
 	}, {
 		// Alter Vschema does not reach the vttablets, so we don't need to run the normalizer test
 		input:                "alter vschema create vindex hash_vdx using `hash`",
@@ -2936,7 +2936,7 @@ var (
 		output: "select json_unquote(@j) from dual",
 	}, {
 		input:  "CREATE TABLE ts (id INT, purchased DATE) PARTITION BY RANGE( YEAR(purchased) ) SUBPARTITION BY HASH( TO_DAYS(purchased) ) ( PARTITION p0 VALUES LESS THAN (1990) (SUBPARTITION s0,SUBPARTITION s1),PARTITION p1 VALUES LESS THAN (2000),PARTITION p2 VALUES LESS THAN MAXVALUE (SUBPARTITION s2,SUBPARTITION s3));",
-		output: "create table ts (\n\tid INT,\n\tpurchased DATE\n) partition by range (YEAR(purchased)) subpartition by hash (TO_DAYS(purchased)) (partition p0 values less than (1990) (subpartition s0, subpartition s1), partition p1 values less than (2000), partition p2 values less than maxvalue (subpartition s2, subpartition `s3`))",
+		output: "create table ts (\n\tid INT,\n\tpurchased DATE\n)\npartition by range (YEAR(purchased)) subpartition by hash (TO_DAYS(purchased))\n(partition p0 values less than (1990) (subpartition s0, subpartition s1),\n partition p1 values less than (2000),\n partition p2 values less than maxvalue (subpartition s2, subpartition `s3`))",
 	}}
 )
 
@@ -4361,7 +4361,12 @@ PARTITION BY RANGE (store_id) (
 	separated DATE not null default '9999-12-31',
 	job_code INT not null,
 	store_id INT not null
-) partition by range (store_id) (partition p0 values less than (6), partition p1 values less than (11), partition p2 values less than (16), partition p3 values less than (21))`,
+)
+partition by range (store_id)
+(partition p0 values less than (6),
+ partition p1 values less than (11),
+ partition p2 values less than (16),
+ partition p3 values less than (21))`,
 		},
 		{
 			input: `CREATE TABLE employees (
@@ -4387,7 +4392,12 @@ PARTITION BY RANGE (store_id) (
 	separated DATE not null default '9999-12-31',
 	job_code INT not null,
 	store_id INT not null
-) partition by range (store_id) (partition p0 values less than (6), partition p1 values less than (11), partition p2 values less than (16), partition p3 values less than maxvalue)`,
+)
+partition by range (store_id)
+(partition p0 values less than (6),
+ partition p1 values less than (11),
+ partition p2 values less than (16),
+ partition p3 values less than maxvalue)`,
 		},
 		{
 			input: `CREATE TABLE employees (
@@ -4412,7 +4422,11 @@ PARTITION BY RANGE (job_code) (
 	separated DATE not null default '9999-12-31',
 	job_code INT not null,
 	store_id INT not null
-) partition by range (job_code) (partition p0 values less than (100), partition p1 values less than (1000), partition p2 values less than (10000))`,
+)
+partition by range (job_code)
+(partition p0 values less than (100),
+ partition p1 values less than (1000),
+ partition p2 values less than (10000))`,
 		},
 		{
 			input: `CREATE TABLE employees (
@@ -4438,7 +4452,12 @@ PARTITION BY RANGE ( YEAR(separated) ) (
 	separated DATE not null default '9999-12-31',
 	job_code INT,
 	store_id INT
-) partition by range (YEAR(separated)) (partition p0 values less than (1991), partition p1 values less than (1996), partition p2 values less than (2001), partition p3 values less than maxvalue)`,
+)
+partition by range (YEAR(separated))
+(partition p0 values less than (1991),
+ partition p1 values less than (1996),
+ partition p2 values less than (2001),
+ partition p3 values less than maxvalue)`,
 		},
 		{
 			input: `CREATE TABLE quarterly_report_status (
@@ -4462,7 +4481,18 @@ PARTITION BY RANGE ( UNIX_TIMESTAMP(report_updated) ) (
 	report_id INT not null,
 	report_status VARCHAR(20) not null,
 	report_updated TIMESTAMP not null default current_timestamp() on update current_timestamp()
-) partition by range (UNIX_TIMESTAMP(report_updated)) (partition p0 values less than (UNIX_TIMESTAMP('2008-01-01 00:00:00')), partition p1 values less than (UNIX_TIMESTAMP('2008-04-01 00:00:00')), partition p2 values less than (UNIX_TIMESTAMP('2008-07-01 00:00:00')), partition p3 values less than (UNIX_TIMESTAMP('2008-10-01 00:00:00')), partition p4 values less than (UNIX_TIMESTAMP('2009-01-01 00:00:00')), partition p5 values less than (UNIX_TIMESTAMP('2009-04-01 00:00:00')), partition p6 values less than (UNIX_TIMESTAMP('2009-07-01 00:00:00')), partition p7 values less than (UNIX_TIMESTAMP('2009-10-01 00:00:00')), partition p8 values less than (UNIX_TIMESTAMP('2010-01-01 00:00:00')), partition p9 values less than maxvalue)`,
+)
+partition by range (UNIX_TIMESTAMP(report_updated))
+(partition p0 values less than (UNIX_TIMESTAMP('2008-01-01 00:00:00')),
+ partition p1 values less than (UNIX_TIMESTAMP('2008-04-01 00:00:00')),
+ partition p2 values less than (UNIX_TIMESTAMP('2008-07-01 00:00:00')),
+ partition p3 values less than (UNIX_TIMESTAMP('2008-10-01 00:00:00')),
+ partition p4 values less than (UNIX_TIMESTAMP('2009-01-01 00:00:00')),
+ partition p5 values less than (UNIX_TIMESTAMP('2009-04-01 00:00:00')),
+ partition p6 values less than (UNIX_TIMESTAMP('2009-07-01 00:00:00')),
+ partition p7 values less than (UNIX_TIMESTAMP('2009-10-01 00:00:00')),
+ partition p8 values less than (UNIX_TIMESTAMP('2010-01-01 00:00:00')),
+ partition p9 values less than maxvalue)`,
 		},
 		{
 			input: `CREATE TABLE quarterly_report_status (
@@ -4486,7 +4516,18 @@ PARTITION BY RANGE ( UNIX_TIMESTAMP(report_updated) ) (
 	report_id INT not null,
 	report_status VARCHAR(20) not null,
 	report_updated TIMESTAMP not null default current_timestamp() on update current_timestamp()
-) partition by range (UNIX_TIMESTAMP(report_updated)) (partition p0 values less than (UNIX_TIMESTAMP('2008-01-01 00:00:00')), partition p1 values less than (UNIX_TIMESTAMP('2008-04-01 00:00:00')), partition p2 values less than (UNIX_TIMESTAMP('2008-07-01 00:00:00')), partition p3 values less than (UNIX_TIMESTAMP('2008-10-01 00:00:00')), partition p4 values less than (UNIX_TIMESTAMP('2009-01-01 00:00:00')), partition p5 values less than (UNIX_TIMESTAMP('2009-04-01 00:00:00')), partition p6 values less than (UNIX_TIMESTAMP('2009-07-01 00:00:00')), partition p7 values less than (UNIX_TIMESTAMP('2009-10-01 00:00:00')), partition p8 values less than (UNIX_TIMESTAMP('2010-01-01 00:00:00')), partition p9 values less than maxvalue)`,
+)
+partition by range (UNIX_TIMESTAMP(report_updated))
+(partition p0 values less than (UNIX_TIMESTAMP('2008-01-01 00:00:00')),
+ partition p1 values less than (UNIX_TIMESTAMP('2008-04-01 00:00:00')),
+ partition p2 values less than (UNIX_TIMESTAMP('2008-07-01 00:00:00')),
+ partition p3 values less than (UNIX_TIMESTAMP('2008-10-01 00:00:00')),
+ partition p4 values less than (UNIX_TIMESTAMP('2009-01-01 00:00:00')),
+ partition p5 values less than (UNIX_TIMESTAMP('2009-04-01 00:00:00')),
+ partition p6 values less than (UNIX_TIMESTAMP('2009-07-01 00:00:00')),
+ partition p7 values less than (UNIX_TIMESTAMP('2009-10-01 00:00:00')),
+ partition p8 values less than (UNIX_TIMESTAMP('2010-01-01 00:00:00')),
+ partition p9 values less than maxvalue)`,
 		},
 		{
 			input: `CREATE TABLE members (
@@ -4509,7 +4550,13 @@ PARTITION BY RANGE COLUMNS(joined) (
 	username VARCHAR(16) not null,
 	email VARCHAR(35),
 	joined DATE not null
-) partition by range columns (joined) (partition p0 values less than ('1960-01-01'), partition p1 values less than ('1970-01-01'), partition p2 values less than ('1980-01-01'), partition p3 values less than ('1990-01-01'), partition p4 values less than maxvalue)`,
+)
+partition by range columns (joined)
+(partition p0 values less than ('1960-01-01'),
+ partition p1 values less than ('1970-01-01'),
+ partition p2 values less than ('1980-01-01'),
+ partition p3 values less than ('1990-01-01'),
+ partition p4 values less than maxvalue)`,
 		},
 		{
 			input: `CREATE TABLE ti (id INT, amount DECIMAL(7,2), tr_date DATE)
@@ -4520,7 +4567,8 @@ PARTITION BY RANGE COLUMNS(joined) (
 	id INT,
 	amount DECIMAL(7,2),
 	tr_date DATE
-) ENGINE INNODB partition by hash (MONTH(tr_date)) partitions 6`,
+) ENGINE INNODB
+partition by hash (MONTH(tr_date)) partitions 6`,
 		},
 		{
 			input: `CREATE TABLE members (
@@ -4538,7 +4586,8 @@ PARTITIONS 6`,
 	username VARCHAR(16) not null,
 	email VARCHAR(35),
 	joined DATE not null
-) partition by key (joined) partitions 6`,
+)
+partition by key (joined) partitions 6`,
 		},
 		{
 			input: `CREATE TABLE t2 (val INT)
@@ -4548,7 +4597,10 @@ PARTITIONS 6`,
 )`,
 			output: `create table t2 (
 	val INT
-) partition by list (val) (partition mypart values in (1, 3, 5), partition MyPart values in (2, 4, 6))`,
+)
+partition by list (val)
+(partition mypart values in (1, 3, 5),
+ partition MyPart values in (2, 4, 6))`,
 		},
 		{
 			input: `CREATE TABLE t2 (val INT)
@@ -4558,7 +4610,10 @@ PARTITIONS 6`,
 )`,
 			output: `create table t2 (
 	val INT
-) partition by list (val) (partition mypart values in (1, 3, 5) storage engine FOOBAR, partition MyPart values in (2, 4, 6))`,
+)
+partition by list (val)
+(partition mypart values in (1, 3, 5) storage engine FOOBAR,
+ partition MyPart values in (2, 4, 6))`,
 		},
 		{
 			input: `CREATE TABLE t2 (val INT)
@@ -4568,7 +4623,10 @@ PARTITIONS 6`,
 )`,
 			output: `create table t2 (
 	val INT
-) partition by list (val) (partition mypart values in (1, 3, 5) engine FOOBAR, partition MyPart values in (2, 4, 6))`,
+)
+partition by list (val)
+(partition mypart values in (1, 3, 5) engine FOOBAR,
+ partition MyPart values in (2, 4, 6))`,
 		},
 		{
 			input: `CREATE TABLE t2 (val INT)
@@ -4578,7 +4636,10 @@ PARTITIONS 6`,
 )`,
 			output: `create table t2 (
 	val INT
-) partition by list (val) (partition mypart values in (1, 3, 5) storage engine FOOBAR, partition MyPart values in (2, 4, 6))`,
+)
+partition by list (val)
+(partition mypart values in (1, 3, 5) storage engine FOOBAR,
+ partition MyPart values in (2, 4, 6))`,
 		},
 		{
 			input: `CREATE TABLE t2 (val INT)
@@ -4588,7 +4649,10 @@ PARTITIONS 6`,
 )`,
 			output: `create table t2 (
 	val INT
-) partition by list (val) (partition mypart values in (1, 3, 5) storage engine FOOBAR comment 'test', partition MyPart values in (2, 4, 6) comment 'test2')`,
+)
+partition by list (val)
+(partition mypart values in (1, 3, 5) storage engine FOOBAR comment 'test',
+ partition MyPart values in (2, 4, 6) comment 'test2')`,
 		},
 		{
 			input: `CREATE TABLE t2 (val INT)
@@ -4598,7 +4662,10 @@ PARTITIONS 6`,
 )`,
 			output: `create table t2 (
 	val INT
-) partition by list (val) (partition mypart values in (1, 3, 5) storage engine FOOBAR data directory 'test', partition MyPart values in (2, 4, 6) data directory 'test2')`,
+)
+partition by list (val)
+(partition mypart values in (1, 3, 5) storage engine FOOBAR data directory 'test',
+ partition MyPart values in (2, 4, 6) data directory 'test2')`,
 		},
 		{
 			input: `CREATE TABLE t2 (val INT)
@@ -4608,13 +4675,19 @@ PARTITIONS 6`,
 )`,
 			output: `create table t2 (
 	val INT
-) partition by list (val) (partition mypart values in (1, 3, 5) index directory 'test', partition MyPart values in (2, 4, 6) index directory 'test2')`,
+)
+partition by list (val)
+(partition mypart values in (1, 3, 5) index directory 'test',
+ partition MyPart values in (2, 4, 6) index directory 'test2')`,
 		},
 		{
 			input: `create table t1 (id int primary key) partition by list (id) (partition p1 values in(11,21), partition p2 values in (12,22))`,
 			output: `create table t1 (
 	id int primary key
-) partition by list (id) (partition p1 values in (11, 21), partition p2 values in (12, 22))`,
+)
+partition by list (id)
+(partition p1 values in (11, 21),
+ partition p2 values in (12, 22))`,
 		},
 		{
 			input: `CREATE TABLE t2 (val INT)
@@ -4624,7 +4697,10 @@ PARTITIONS 6`,
 )`,
 			output: `create table t2 (
 	val INT
-) partition by list (val) (partition mypart values in (1, 3, 5) storage engine FOOBAR comment 'before' data directory 'test', partition MyPart values in (2, 4, 6) data directory 'test2')`,
+)
+partition by list (val)
+(partition mypart values in (1, 3, 5) storage engine FOOBAR comment 'before' data directory 'test',
+ partition MyPart values in (2, 4, 6) data directory 'test2')`,
 		},
 		{
 			input: `CREATE TABLE t2 (val INT)
@@ -4634,7 +4710,10 @@ PARTITIONS 6`,
 )`,
 			output: `create table t2 (
 	val INT
-) partition by list (val) (partition mypart values in (1, 3, 5) max_rows 4, partition MyPart values in (2, 4, 6) max_rows 10)`,
+)
+partition by list (val)
+(partition mypart values in (1, 3, 5) max_rows 4,
+ partition MyPart values in (2, 4, 6) max_rows 10)`,
 		},
 		{
 			input: `CREATE TABLE t2 (val INT)
@@ -4644,7 +4723,10 @@ PARTITIONS 6`,
 )`,
 			output: `create table t2 (
 	val INT
-) partition by list (val) (partition mypart values in (1, 3, 5) min_rows 4, partition MyPart values in (2, 4, 6) min_rows 10)`,
+)
+partition by list (val)
+(partition mypart values in (1, 3, 5) min_rows 4,
+ partition MyPart values in (2, 4, 6) min_rows 10)`,
 		},
 		{
 			input: `CREATE TABLE t2 (val INT)
@@ -4654,7 +4736,10 @@ PARTITIONS 6`,
 )`,
 			output: `create table t2 (
 	val INT
-) partition by list (val) (partition mypart values in (1, 3, 5) tablespace innodb_system, partition MyPart values in (2, 4, 6) tablespace innodb_system)`,
+)
+partition by list (val)
+(partition mypart values in (1, 3, 5) tablespace innodb_system,
+ partition MyPart values in (2, 4, 6) tablespace innodb_system)`,
 		},
 		{
 			// index with an expression
@@ -4746,7 +4831,11 @@ PARTITIONS 6`,
 			output: `create table ts (
 	id INT,
 	purchased DATE
-) partition by range (YEAR(purchased)) subpartition by hash (TO_DAYS(purchased)) (partition p0 values less than (1990) (subpartition s0, subpartition s1), partition p1 values less than (2000) (subpartition s2, subpartition s31), partition p2 values less than maxvalue (subpartition s4, subpartition s5))`,
+)
+partition by range (YEAR(purchased)) subpartition by hash (TO_DAYS(purchased))
+(partition p0 values less than (1990) (subpartition s0, subpartition s1),
+ partition p1 values less than (2000) (subpartition s2, subpartition s31),
+ partition p2 values less than maxvalue (subpartition s4, subpartition s5))`,
 		},
 		{
 			input: `CREATE TABLE ts (id INT, purchased DATE)
@@ -4768,7 +4857,11 @@ PARTITIONS 6`,
 			output: `create table ts (
 	id INT,
 	purchased DATE
-) partition by range (YEAR(purchased)) subpartition by hash (TO_DAYS(purchased)) (partition p0 values less than (1990) (subpartition s0 storage engine innodb data directory '/data', subpartition s1 comment 'this is s1' index directory '/index'), partition p1 values less than (2000) (subpartition s2 max_rows 4, subpartition s31 min_rows 5 tablespace t2), partition p2 values less than maxvalue (subpartition s4, subpartition s5))`,
+)
+partition by range (YEAR(purchased)) subpartition by hash (TO_DAYS(purchased))
+(partition p0 values less than (1990) (subpartition s0 storage engine innodb data directory '/data', subpartition s1 comment 'this is s1' index directory '/index'),
+ partition p1 values less than (2000) (subpartition s2 max_rows 4, subpartition s31 min_rows 5 tablespace t2),
+ partition p2 values less than maxvalue (subpartition s4, subpartition s5))`,
 		},
 		{
 			input:  "create table t (i bigint) charset ascii",
@@ -5109,7 +5202,10 @@ func TestParseVersionedComments(t *testing.T) {
 			output: `create table table1 (
 	id int
 ) ENGINE InnoDB,
-  CHARSET utf8mb4 partition by range (id) (partition x values less than (5) engine InnoDB, partition t values less than (20) engine InnoDB)`,
+  CHARSET utf8mb4
+partition by range (id)
+(partition x values less than (5) engine InnoDB,
+ partition t values less than (20) engine InnoDB)`,
 		},
 	}
 
diff --git a/go/vt/sqlparser/tracked_buffer_test.go b/go/vt/sqlparser/tracked_buffer_test.go
index cd59fe7f83..6d3bdb0795 100644
--- a/go/vt/sqlparser/tracked_buffer_test.go
+++ b/go/vt/sqlparser/tracked_buffer_test.go
@@ -106,7 +106,7 @@ func TestCanonicalOutput(t *testing.T) {
 		},
 		{
 			"create table a (id int not null primary key) engine InnoDB, charset utf8mb4, collate utf8mb4_0900_ai_ci partition by range (`id`) (partition `p10` values less than(10) engine InnoDB)",
-			"CREATE TABLE `a` (\n\t`id` int NOT NULL PRIMARY KEY\n) ENGINE InnoDB,\n  CHARSET utf8mb4,\n  COLLATE utf8mb4_0900_ai_ci PARTITION BY RANGE (`id`) (PARTITION `p10` VALUES LESS THAN (10) ENGINE InnoDB)",
+			"CREATE TABLE `a` (\n\t`id` int NOT NULL PRIMARY KEY\n) ENGINE InnoDB,\n  CHARSET utf8mb4,\n  COLLATE utf8mb4_0900_ai_ci\nPARTITION BY RANGE (`id`)\n(PARTITION `p10` VALUES LESS THAN (10) ENGINE InnoDB)",
 		},
 		{
 			"alter table a comment='a b c'",
@@ -142,11 +142,11 @@ func TestCanonicalOutput(t *testing.T) {
 		},
 		{
 			"alter table t1 partition by hash (id) partitions 5",
-			"ALTER TABLE `t1` PARTITION BY HASH (`id`) PARTITIONS 5",
+			"ALTER TABLE `t1` \nPARTITION BY HASH (`id`) PARTITIONS 5",
 		},
 		{
 			"alter table t1 partition by list (id) (partition p1 values in (11, 21), partition p2 values in (12, 22))",
-			"ALTER TABLE `t1` PARTITION BY LIST (`id`) (PARTITION `p1` VALUES IN (11, 21), PARTITION `p2` VALUES IN (12, 22))",
+			"ALTER TABLE `t1` \nPARTITION BY LIST (`id`)\n(PARTITION `p1` VALUES IN (11, 21),\n PARTITION `p2` VALUES IN (12, 22))",
 		},
 		{
 			"alter table t1 row_format=compressed, character set=utf8",
diff --git a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
index d147c64730..13df2feb0e 100644
--- a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
@@ -2298,7 +2298,7 @@ Gen4 plan same as above
       {
         "OperatorType": "Projection",
         "Expressions": [
-          "[COLUMN 0]",
+          "[COLUMN 0] as a",
           "[COLUMN 1]"
         ],
         "Inputs": [
@@ -3240,7 +3240,7 @@ Gen4 error: cannot push projections in ordered aggregates
       {
         "OperatorType": "Projection",
         "Expressions": [
-          "[COLUMN 0]",
+          "[COLUMN 0] as a",
           "[COLUMN 2] * [COLUMN 3] as count(*)",
           "[COLUMN 1]"
         ],
@@ -3298,7 +3298,7 @@ Gen4 error: cannot push projections in ordered aggregates
       {
         "OperatorType": "Projection",
         "Expressions": [
-          "[COLUMN 0]",
+          "[COLUMN 0] as a",
           "[COLUMN 2] * [COLUMN 3] as count(user_extra.a)",
           "[COLUMN 1]"
         ],
@@ -3358,7 +3358,7 @@ Gen4 error: cannot push projections in ordered aggregates
         "Expressions": [
           "([COLUMN 2] * [COLUMN 3]) * [COLUMN 4] as count(u.textcol1)",
           "([COLUMN 5] * [COLUMN 6]) * [COLUMN 7] as count(ue.foo)",
-          "[COLUMN 0]",
+          "[COLUMN 0] as bar",
           "[COLUMN 1]"
         ],
         "Inputs": [
@@ -3717,8 +3717,8 @@ Gen4 error: cannot push projections in ordered aggregates
       {
         "OperatorType": "Projection",
         "Expressions": [
-          "[COLUMN 0]",
-          "[COLUMN 1]",
+          "[COLUMN 0] as textcol1",
+          "[COLUMN 1] as val2",
           "[COLUMN 2]"
         ],
         "Inputs": [
@@ -3808,9 +3808,9 @@ Gen4 error: cannot push projections in ordered aggregates
       {
         "OperatorType": "Projection",
         "Expressions": [
-          "[COLUMN 0]",
+          "[COLUMN 0] as col",
           "[COLUMN 3] as min(user_extra.foo)",
-          "[COLUMN 1]",
+          "[COLUMN 1] as bar",
           "[COLUMN 4] as max(user_extra.bar)",
           "[COLUMN 2]"
         ],
diff --git a/go/vt/vtgate/planbuilder/testdata/from_cases.txt b/go/vt/vtgate/planbuilder/testdata/from_cases.txt
index 96860e0964..72510897cf 100644
--- a/go/vt/vtgate/planbuilder/testdata/from_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/from_cases.txt
@@ -4667,3 +4667,44 @@ Gen4 plan same as above
     ]
   }
 }
+
+# join with USING construct
+"select * from authoritative join unsharded_authoritative using(col1)"
+"unsupported: join with USING(column_list) clause for complex queries"
+{
+  "QueryType": "SELECT",
+  "Original": "select * from authoritative join unsharded_authoritative using(col1)",
+  "Instructions": {
+    "OperatorType": "Join",
+    "Variant": "Join",
+    "JoinColumnIndexes": "L:1,L:2,L:3,R:0",
+    "JoinVars": {
+      "authoritative_col1": 0
+    },
+    "TableName": "authoritative_unsharded_authoritative",
+    "Inputs": [
+      {
+        "OperatorType": "Route",
+        "Variant": "Scatter",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select authoritative.col1, authoritative.col1 as col1, authoritative.user_id as user_id, authoritative.col2 as col2 from authoritative where 1 != 1",
+        "Query": "select authoritative.col1, authoritative.col1 as col1, authoritative.user_id as user_id, authoritative.col2 as col2 from authoritative",
+        "Table": "authoritative"
+      },
+      {
+        "OperatorType": "Route",
+        "Variant": "Unsharded",
+        "Keyspace": {
+          "Name": "main",
+          "Sharded": false
+        },
+        "FieldQuery": "select unsharded_authoritative.col2 as col2 from unsharded_authoritative where 1 != 1",
+        "Query": "select unsharded_authoritative.col2 as col2 from unsharded_authoritative where unsharded_authoritative.col1 = :authoritative_col1",
+        "Table": "unsharded_authoritative"
+      }
+    ]
+  }
+}
diff --git a/go/vt/vtgate/planbuilder/testdata/systemtables_cases.txt b/go/vt/vtgate/planbuilder/testdata/systemtables_cases.txt
index e16280edf1..86029491aa 100644
--- a/go/vt/vtgate/planbuilder/testdata/systemtables_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/systemtables_cases.txt
@@ -426,7 +426,22 @@ Gen4 plan same as above
     "Table": "information_schema.referential_constraints, information_schema.key_column_usage"
   }
 }
-Gen4 error: unsupported: join with USING(column_list) clause for complex queries
+{
+  "QueryType": "SELECT",
+  "Original": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as name, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc join information_schema.key_column_usage as fk using (constraint_schema, constraint_name) where fk.referenced_column_name is not null and fk.table_schema = database() and fk.table_name = ':vtg1' and rc.constraint_schema = database() and rc.table_name = ':vtg1'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "DBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc, information_schema.key_column_usage as fk where 1 != 1",
+    "Query": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc, information_schema.key_column_usage as fk where rc.constraint_schema = database() and rc.table_name = :rc_table_name and fk.referenced_column_name is not null and fk.table_schema = database() and fk.table_name = :fk_table_name",
+    "SysTableTableName": "[fk_table_name:VARCHAR(\":vtg1\"), rc_table_name:VARCHAR(\":vtg1\")]",
+    "Table": "information_schema.key_column_usage, information_schema.referential_constraints"
+  }
+}
 
 #rails_query 2
 "SELECT * FROM information_schema.schemata WHERE schema_name = 'user'"
@@ -488,7 +503,23 @@ Gen4 plan same as above
     "Table": "information_schema.referential_constraints, information_schema.key_column_usage"
   }
 }
-Gen4 error: unsupported: join with USING(column_list) clause for complex queries
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT fk.referenced_table_name AS 'to_table', fk.referenced_column_name AS 'primary_key',fk.column_name AS 'column',fk.constraint_name AS 'name',rc.update_rule AS 'on_update',rc.delete_rule AS 'on_delete' FROM information_schema.referential_constraints rc JOIN information_schema.key_column_usage fk USING (constraint_schema, constraint_name) WHERE fk.referenced_column_name IS NOT NULL AND fk.table_schema = 'table_schema' AND fk.table_name = 'table_name' AND rc.constraint_schema = 'table_schema' AND rc.table_name = 'table_name'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "DBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc, information_schema.key_column_usage as fk where 1 != 1",
+    "Query": "select fk.referenced_table_name as to_table, fk.referenced_column_name as primary_key, fk.column_name as `column`, fk.constraint_name as `name`, rc.update_rule as on_update, rc.delete_rule as on_delete from information_schema.referential_constraints as rc, information_schema.key_column_usage as fk where rc.constraint_schema = :__vtschemaname and rc.table_name = :rc_table_name and fk.referenced_column_name is not null and fk.table_schema = :__vtschemaname and fk.table_name = :fk_table_name",
+    "SysTableTableName": "[fk_table_name:VARCHAR(\"table_name\"), rc_table_name:VARCHAR(\"table_name\")]",
+    "SysTableTableSchema": "[VARCHAR(\"table_schema\"), VARCHAR(\"table_schema\")]",
+    "Table": "information_schema.key_column_usage, information_schema.referential_constraints"
+  }
+}
 
 #rails_query 5
 "SELECT cc.constraint_name AS 'name', cc.check_clause AS 'expression' FROM information_schema.check_constraints cc JOIN information_schema.table_constraints tc USING (constraint_schema, constraint_name) WHERE tc.table_schema = 'table_schema' AND tc.table_name = 'table_name' AND cc.constraint_schema = 'constraint_schema'"
@@ -509,7 +540,23 @@ Gen4 error: unsupported: join with USING(column_list) clause for complex queries
     "Table": "information_schema.check_constraints, information_schema.table_constraints"
   }
 }
-Gen4 error: unsupported: join with USING(column_list) clause for complex queries
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT cc.constraint_name AS 'name', cc.check_clause AS 'expression' FROM information_schema.check_constraints cc JOIN information_schema.table_constraints tc USING (constraint_schema, constraint_name) WHERE tc.table_schema = 'table_schema' AND tc.table_name = 'table_name' AND cc.constraint_schema = 'constraint_schema'",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "DBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select cc.constraint_name as `name`, cc.check_clause as expression from information_schema.check_constraints as cc, information_schema.table_constraints as tc where 1 != 1",
+    "Query": "select cc.constraint_name as `name`, cc.check_clause as expression from information_schema.check_constraints as cc, information_schema.table_constraints as tc where cc.constraint_schema = :__vtschemaname and tc.table_schema = :__vtschemaname and tc.table_name = :tc_table_name",
+    "SysTableTableName": "[tc_table_name:VARCHAR(\"table_name\")]",
+    "SysTableTableSchema": "[VARCHAR(\"constraint_schema\"), VARCHAR(\"table_schema\")]",
+    "Table": "information_schema.check_constraints, information_schema.table_constraints"
+  }
+}
 
 #rails_query 6
 "SELECT column_name FROM information_schema.statistics WHERE index_name = 'PRIMARY' AND table_schema = 'table_schema' AND table_name = 'table_name' ORDER BY seq_in_index"
diff --git a/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt b/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt
index c066ee6b57..d4d36a623c 100644
--- a/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt
@@ -33,10 +33,10 @@ Gen4 error: unsupported: cross-shard correlated subquery
               {
                 "OperatorType": "Projection",
                 "Expressions": [
-                  "[COLUMN 0]",
+                  "[COLUMN 0] as l_orderkey",
                   "([COLUMN 6] * [COLUMN 7]) * [COLUMN 8] as revenue",
-                  "[COLUMN 1]",
-                  "[COLUMN 2]",
+                  "[COLUMN 1] as o_orderdate",
+                  "[COLUMN 2] as o_shippriority",
                   "[COLUMN 5]",
                   "[COLUMN 4]",
                   "[COLUMN 3]"
@@ -138,7 +138,7 @@ Gen4 error: unsupported: cross-shard correlated subquery
       {
         "OperatorType": "Projection",
         "Expressions": [
-          "[COLUMN 1]",
+          "[COLUMN 1] as o_orderpriority",
           "[COLUMN 2] as order_count",
           "[COLUMN 3]"
         ],
@@ -206,7 +206,7 @@ Gen4 error: unsupported: cross-shard correlated subquery
           {
             "OperatorType": "Projection",
             "Expressions": [
-              "[COLUMN 0]",
+              "[COLUMN 0] as n_name",
               "(((([COLUMN 2] * [COLUMN 3]) * [COLUMN 4]) * [COLUMN 5]) * [COLUMN 6]) * [COLUMN 7] as revenue",
               "[COLUMN 1]"
             ],
@@ -431,9 +431,9 @@ Gen4 error: unsupported: cross-shard correlated subquery
       {
         "OperatorType": "Projection",
         "Expressions": [
-          "[COLUMN 4]",
-          "[COLUMN 5]",
-          "[COLUMN 6]",
+          "[COLUMN 4] as supp_nation",
+          "[COLUMN 5] as cust_nation",
+          "[COLUMN 6] as l_year",
           "(((([COLUMN 10] * [COLUMN 11]) * [COLUMN 12]) * [COLUMN 13]) * [COLUMN 14]) * [COLUMN 15] as revenue",
           "[COLUMN 9]",
           "[COLUMN 8]",
@@ -630,14 +630,14 @@ Gen4 error: aggregation on columns from different sources not supported yet
               {
                 "OperatorType": "Projection",
                 "Expressions": [
-                  "[COLUMN 0]",
-                  "[COLUMN 1]",
+                  "[COLUMN 0] as c_custkey",
+                  "[COLUMN 1] as c_name",
                   "(([COLUMN 14] * [COLUMN 15]) * [COLUMN 16]) * [COLUMN 17] as revenue",
-                  "[COLUMN 2]",
-                  "[COLUMN 4]",
-                  "[COLUMN 5]",
-                  "[COLUMN 3]",
-                  "[COLUMN 6]",
+                  "[COLUMN 2] as c_acctbal",
+                  "[COLUMN 4] as n_name",
+                  "[COLUMN 5] as c_address",
+                  "[COLUMN 3] as c_phone",
+                  "[COLUMN 6] as c_comment",
                   "[COLUMN 13]",
                   "[COLUMN 12]",
                   "[COLUMN 11]",
@@ -774,7 +774,7 @@ Gen4 error: unsupported: in scatter query: complex aggregate expression
       {
         "OperatorType": "Projection",
         "Expressions": [
-          "[COLUMN 0]",
+          "[COLUMN 0] as l_shipmode",
           "[COLUMN 2] * [COLUMN 3] as high_line_count",
           "[COLUMN 4] * [COLUMN 5] as low_line_count",
           "[COLUMN 1]"
@@ -1067,7 +1067,7 @@ Gen4 error: unsupported: cross-shard correlated subquery
               {
                 "OperatorType": "Projection",
                 "Expressions": [
-                  "[COLUMN 0]",
+                  "[COLUMN 0] as s_name",
                   "(([COLUMN 2] * [COLUMN 3]) * [COLUMN 4]) * [COLUMN 5] as numwait",
                   "[COLUMN 1]"
                 ],
diff --git a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
index ad8903d62a..03142ae620 100644
--- a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
@@ -41,12 +41,12 @@ Gen4 plan same as above
 # join with USING construct
 "select * from user join user_extra using(id)"
 "unsupported: join with USING(column_list) clause for complex queries"
-Gen4 plan same as above
+Gen4 error: can't handle JOIN USING without authoritative tables
 
 # join with USING construct with 3 tables
 "select user.id from user join user_extra using(id) join music using(id2)"
 "unsupported: join with USING(column_list) clause for complex queries"
-Gen4 plan same as above
+Gen4 error: can't handle JOIN USING without authoritative tables
 
 # natural left join
 "select * from user natural left join user_extra"
@@ -522,11 +522,6 @@ Gen4 error: can't push predicates on concatenate
 "expr cannot be translated, not supported: (select 1 from `user` where id = 1)"
 Gen4 plan same as above
 
-# join with USING construct
-"select user.id from user join user_extra using(id)"
-"unsupported: join with USING(column_list) clause for complex queries"
-Gen4 plan same as above
-
 # lateral derived tables
 "select * from user, lateral (select * from user_extra where user_id = user.id) t"
 "unsupported: lateral derived tables"
diff --git a/go/vt/vtgate/planbuilder/testdata/vindex_func_cases.txt b/go/vt/vtgate/planbuilder/testdata/vindex_func_cases.txt
index c2457d078a..a420227a5b 100644
--- a/go/vt/vtgate/planbuilder/testdata/vindex_func_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/vindex_func_cases.txt
@@ -146,41 +146,7 @@ Gen4 plan same as above
     ]
   }
 }
-{
-  "QueryType": "SELECT",
-  "Original": "select user_index.keyspace_id, unsharded.id from user_index join unsharded where user_index.id = :id",
-  "Instructions": {
-    "OperatorType": "Join",
-    "Variant": "Join",
-    "JoinColumnIndexes": "L:0,R:0",
-    "TableName": "_unsharded",
-    "Inputs": [
-      {
-        "OperatorType": "VindexFunc",
-        "Variant": "VindexMap",
-        "Columns": [
-          1
-        ],
-        "Fields": {
-          "user_index.keyspace_id": "VARBINARY"
-        },
-        "Value": ":id",
-        "Vindex": "user_index"
-      },
-      {
-        "OperatorType": "Route",
-        "Variant": "Unsharded",
-        "Keyspace": {
-          "Name": "main",
-          "Sharded": false
-        },
-        "FieldQuery": "select unsharded.id from unsharded where 1 != 1",
-        "Query": "select unsharded.id from unsharded",
-        "Table": "unsharded"
-      }
-    ]
-  }
-}
+Gen4 plan same as above
 
 # Join vindexFunc on RHS
 "select user_index.keyspace_id, unsharded.id from unsharded join user_index where user_index.id = :id"
@@ -219,41 +185,7 @@ Gen4 plan same as above
     ]
   }
 }
-{
-  "QueryType": "SELECT",
-  "Original": "select user_index.keyspace_id, unsharded.id from unsharded join user_index where user_index.id = :id",
-  "Instructions": {
-    "OperatorType": "Join",
-    "Variant": "Join",
-    "JoinColumnIndexes": "R:0,L:0",
-    "TableName": "unsharded_",
-    "Inputs": [
-      {
-        "OperatorType": "Route",
-        "Variant": "Unsharded",
-        "Keyspace": {
-          "Name": "main",
-          "Sharded": false
-        },
-        "FieldQuery": "select unsharded.id from unsharded where 1 != 1",
-        "Query": "select unsharded.id from unsharded",
-        "Table": "unsharded"
-      },
-      {
-        "OperatorType": "VindexFunc",
-        "Variant": "VindexMap",
-        "Columns": [
-          1
-        ],
-        "Fields": {
-          "user_index.keyspace_id": "VARBINARY"
-        },
-        "Value": ":id",
-        "Vindex": "user_index"
-      }
-    ]
-  }
-}
+Gen4 plan same as above
 
 # Join with vindexFunc on a column of it, already present in select list
 "select user_index.id, user_index.keyspace_id, unsharded.id from user_index join unsharded where user_index.id = :id and unsharded.id = user_index.id"
@@ -297,46 +229,7 @@ Gen4 plan same as above
     ]
   }
 }
-{
-  "QueryType": "SELECT",
-  "Original": "select user_index.id, user_index.keyspace_id, unsharded.id from user_index join unsharded where user_index.id = :id and unsharded.id = user_index.id",
-  "Instructions": {
-    "OperatorType": "Join",
-    "Variant": "Join",
-    "JoinColumnIndexes": "L:0,L:1,R:0",
-    "JoinVars": {
-      "user_index_id": 0
-    },
-    "TableName": "_unsharded",
-    "Inputs": [
-      {
-        "OperatorType": "VindexFunc",
-        "Variant": "VindexMap",
-        "Columns": [
-          0,
-          1
-        ],
-        "Fields": {
-          "user_index.id": "VARBINARY",
-          "user_index.keyspace_id": "VARBINARY"
-        },
-        "Value": ":id",
-        "Vindex": "user_index"
-      },
-      {
-        "OperatorType": "Route",
-        "Variant": "Unsharded",
-        "Keyspace": {
-          "Name": "main",
-          "Sharded": false
-        },
-        "FieldQuery": "select unsharded.id from unsharded where 1 != 1",
-        "Query": "select unsharded.id from unsharded where unsharded.id = :user_index_id",
-        "Table": "unsharded"
-      }
-    ]
-  }
-}
+Gen4 plan same as above
 
 # Join with vindexFunc on a column of it, already present at the end of the select list
 "select user_index.keyspace_id, user_index.id, unsharded.id from user_index join unsharded where user_index.id = :id and unsharded.id = user_index.id"
@@ -400,8 +293,8 @@ Gen4 plan same as above
           1
         ],
         "Fields": {
-          "user_index.id": "VARBINARY",
-          "user_index.keyspace_id": "VARBINARY"
+          "id": "VARBINARY",
+          "keyspace_id": "VARBINARY"
         },
         "Value": ":id",
         "Vindex": "user_index"
@@ -483,8 +376,8 @@ Gen4 plan same as above
           1
         ],
         "Fields": {
-          "user_index.id": "VARBINARY",
-          "user_index.keyspace_id": "VARBINARY"
+          "id": "VARBINARY",
+          "keyspace_id": "VARBINARY"
         },
         "Value": ":id",
         "Vindex": "user_index"
@@ -566,8 +459,8 @@ Gen4 plan same as above
           1
         ],
         "Fields": {
-          "ui.id": "VARBINARY",
-          "ui.keyspace_id": "VARBINARY"
+          "id": "VARBINARY",
+          "keyspace_id": "VARBINARY"
         },
         "Value": ":id",
         "Vindex": "user_index"
diff --git a/go/vt/vtgate/semantics/early_rewriter_test.go b/go/vt/vtgate/semantics/early_rewriter_test.go
index 7a3902bfd0..91d5387f49 100644
--- a/go/vt/vtgate/semantics/early_rewriter_test.go
+++ b/go/vt/vtgate/semantics/early_rewriter_test.go
@@ -63,6 +63,28 @@ func TestExpandStar(t *testing.T) {
 				}},
 				ColumnListAuthoritative: false,
 			},
+			"t4": {
+				Name: sqlparser.NewTableIdent("t4"),
+				Columns: []vindexes.Column{{
+					Name: sqlparser.NewColIdent("c1"),
+					Type: sqltypes.VarChar,
+				}, {
+					Name: sqlparser.NewColIdent("c4"),
+					Type: sqltypes.VarChar,
+				}},
+				ColumnListAuthoritative: true,
+			},
+			"t5": {
+				Name: sqlparser.NewTableIdent("t5"),
+				Columns: []vindexes.Column{{
+					Name: sqlparser.NewColIdent("a"),
+					Type: sqltypes.VarChar,
+				}, {
+					Name: sqlparser.NewColIdent("b"),
+					Type: sqltypes.VarChar,
+				}},
+				ColumnListAuthoritative: true,
+			},
 		},
 	}
 	cDB := "db"
@@ -106,6 +128,108 @@ func TestExpandStar(t *testing.T) {
 	}, {
 		sql:    "select foo.* from t1, t2",
 		expErr: "Unknown table 'foo'",
+	}, {
+		sql:    "select * from t1 join t2 on t1.a = t2.c1",
+		expSQL: "select t1.a as a, t1.b as b, t1.c as c, t2.c1 as c1, t2.c2 as c2 from t1 join t2 on t1.a = t2.c1",
+	}, {
+		sql:    "select * from t2 join t4 using (c1)",
+		expSQL: "select t2.c1 as c1, t2.c2 as c2, t4.c4 as c4 from t2 join t4 where t2.c1 = t4.c1",
+	}, {
+		sql:    "select * from t2 join t4 using (c1) join t2 as X using (c1)",
+		expSQL: "select t2.c1 as c1, t2.c2 as c2, t4.c4 as c4, X.c2 as c2 from t2 join t4 join t2 as X where t2.c1 = t4.c1 and t2.c1 = X.c1 and t4.c1 = X.c1",
+	}, {
+		sql:    "select * from t2 join t4 using (c1), t2 as t2b join t4 as t4b using (c1)",
+		expSQL: "select t2.c1 as c1, t2.c2 as c2, t4.c4 as c4, t2b.c1 as c1, t2b.c2 as c2, t4b.c4 as c4 from t2 join t4, t2 as t2b join t4 as t4b where t2b.c1 = t4b.c1 and t2.c1 = t4.c1",
+	}, {
+		sql:    "select * from t1 join t5 using (b)",
+		expSQL: "select t1.b as b, t1.a as a, t1.c as c, t5.a as a from t1 join t5 where t1.b = t5.b",
+	}, {
+		sql:    "select * from t1 join t5 using (b) having b = 12",
+		expSQL: "select t1.b as b, t1.a as a, t1.c as c, t5.a as a from t1 join t5 where t1.b = t5.b having b = 12",
+	}, {
+		sql:    "select 1 from t1 join t5 using (b) having b = 12",
+		expSQL: "select 1 from t1 join t5 where t1.b = t5.b having t1.b = 12",
+	}}
+	for _, tcase := range tcases {
+		t.Run(tcase.sql, func(t *testing.T) {
+			ast, err := sqlparser.Parse(tcase.sql)
+			require.NoError(t, err)
+			selectStatement, isSelectStatement := ast.(*sqlparser.Select)
+			require.True(t, isSelectStatement, "analyzer expects a select statement")
+			st, err := Analyze(selectStatement, cDB, schemaInfo)
+			if tcase.expErr == "" {
+				require.NoError(t, err)
+				require.NoError(t, st.NotUnshardedErr)
+				require.NoError(t, st.NotSingleRouteErr)
+				assert.Equal(t, tcase.expSQL, sqlparser.String(selectStatement))
+			} else {
+				require.EqualError(t, err, tcase.expErr)
+			}
+		})
+	}
+}
+
+func TestRewriteJoinUsingColumns(t *testing.T) {
+	schemaInfo := &FakeSI{
+		Tables: map[string]*vindexes.Table{
+			"t1": {
+				Name: sqlparser.NewTableIdent("t1"),
+				Columns: []vindexes.Column{{
+					Name: sqlparser.NewColIdent("a"),
+					Type: sqltypes.VarChar,
+				}, {
+					Name: sqlparser.NewColIdent("b"),
+					Type: sqltypes.VarChar,
+				}, {
+					Name: sqlparser.NewColIdent("c"),
+					Type: sqltypes.VarChar,
+				}},
+				ColumnListAuthoritative: true,
+			},
+			"t2": {
+				Name: sqlparser.NewTableIdent("t2"),
+				Columns: []vindexes.Column{{
+					Name: sqlparser.NewColIdent("a"),
+					Type: sqltypes.VarChar,
+				}, {
+					Name: sqlparser.NewColIdent("b"),
+					Type: sqltypes.VarChar,
+				}, {
+					Name: sqlparser.NewColIdent("c"),
+					Type: sqltypes.VarChar,
+				}},
+				ColumnListAuthoritative: true,
+			},
+			"t3": {
+				Name: sqlparser.NewTableIdent("t3"),
+				Columns: []vindexes.Column{{
+					Name: sqlparser.NewColIdent("a"),
+					Type: sqltypes.VarChar,
+				}, {
+					Name: sqlparser.NewColIdent("b"),
+					Type: sqltypes.VarChar,
+				}, {
+					Name: sqlparser.NewColIdent("c"),
+					Type: sqltypes.VarChar,
+				}},
+				ColumnListAuthoritative: true,
+			},
+		},
+	}
+	cDB := "db"
+	tcases := []struct {
+		sql    string
+		expSQL string
+		expErr string
+	}{{
+		sql:    "select 1 from t1 join t2 using (a) where a = 42",
+		expSQL: "select 1 from t1 join t2 where t1.a = t2.a and t1.a = 42",
+	}, {
+		sql:    "select 1 from t1 join t2 using (a), t3 where a = 42",
+		expErr: "Column 'a' in field list is ambiguous",
+	}, {
+		sql:    "select 1 from t1 join t2 using (a), t1 as b join t3 on (a) where a = 42",
+		expErr: "Column 'a' in field list is ambiguous",
 	}}
 	for _, tcase := range tcases {
 		t.Run(tcase.sql, func(t *testing.T) {
@@ -122,6 +246,7 @@ func TestExpandStar(t *testing.T) {
 			}
 		})
 	}
+
 }
 
 func TestOrderByGroupByLiteral(t *testing.T) {
diff --git a/go/vt/vtgate/semantics/tabletset_test.go b/go/vt/vtgate/semantics/tabletset_test.go
index 300810e18a..50e437220b 100644
--- a/go/vt/vtgate/semantics/tabletset_test.go
+++ b/go/vt/vtgate/semantics/tabletset_test.go
@@ -48,10 +48,10 @@ func TestTableSet_IsSolvedBy(t *testing.T) {
 }
 
 func TestTableSet_Constituents(t *testing.T) {
-	assert.Equal(t, []int{0, 1, 2}, (F123).Constituents())
-	assert.Equal(t, []int{0, 1}, (F12).Constituents())
-	assert.Equal(t, []int{0, 2}, (F1.Merge(F3)).Constituents())
-	assert.Equal(t, []int{1, 2}, (F2.Merge(F3)).Constituents())
+	assert.Equal(t, []TableSet{F1, F2, F3}, (F123).Constituents())
+	assert.Equal(t, []TableSet{F1, F2}, (F12).Constituents())
+	assert.Equal(t, []TableSet{F1, F3}, (F1.Merge(F3)).Constituents())
+	assert.Equal(t, []TableSet{F2, F3}, (F2.Merge(F3)).Constituents())
 	assert.Empty(t, TableSet{}.Constituents())
 }
 
@@ -65,12 +65,12 @@ func TestTableSet_LargeTablesConstituents(t *testing.T) {
 	const GapSize = 32
 
 	var ts TableSet
-	var expected []int
+	var expected []TableSet
 	var table int
 
 	for t := 0; t < 256; t++ {
 		table += rand.Intn(GapSize) + 1
-		expected = append(expected, table)
+		expected = append(expected, SingleTableSet(table))
 		ts.AddTable(table)
 	}
 
@@ -97,9 +97,9 @@ func TestTabletSet_LargeMergeInPlace(t *testing.T) {
 		result.MergeInPlace(ts)
 	}
 
-	var expected = make([]int, SetRange*Blocks)
+	var expected = make([]TableSet, SetRange*Blocks)
 	for tid := range expected {
-		expected[tid] = tid
+		expected[tid] = SingleTableSet(tid)
 	}
 
 	assert.Equal(t, expected, result.Constituents())
@@ -125,9 +125,9 @@ func TestTabletSet_LargeMerge(t *testing.T) {
 		result = result.Merge(ts)
 	}
 
-	var expected = make([]int, SetRange*Blocks)
+	var expected = make([]TableSet, SetRange*Blocks)
 	for tid := range expected {
-		expected[tid] = tid
+		expected[tid] = SingleTableSet(tid)
 	}
 
 	assert.Equal(t, expected, result.Constituents())