Enable vschema operations via vtgate in vttestserver

go/cmd/vttestserver/main.go

@@ -172,6 +172,8 @@ func parseFlags() (config vttest.Config, env vttest.Environment, err error) {
 
 	flag.BoolVar(&config.InitWorkflowManager, "workflow_manager_init", false, "Enable workflow manager")
 
+	flag.StringVar(&config.VSchemaDDLAuthorizedUsers, "vschema_ddl_authorized_users", "", "Comma separated list of users authorized to execute vschema ddl operations via vtgate")
+
 	flag.Parse()
 
 	if basePort != 0 {

go/cmd/vttestserver/vttestserver_test.go

@@ -22,6 +22,9 @@ import (
 	"os"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+	"vitess.io/vitess/go/mysql"
+
 	"vitess.io/vitess/go/vt/vttest"
 
 	"github.com/golang/protobuf/jsonpb"
@@ -45,14 +48,37 @@ func TestRunsVschemaMigrations(t *testing.T) {
 	tabletHostname := "-tablet_hostname=localhost"
 	keyspaceArg := "-keyspaces=test_keyspace,app_customer"
 	numShardsArg := "-num_shards=2,2"
+	vschemaDDLAuthorizedUsers := "-vschema_ddl_authorized_users=%"
 
-	os.Args = append(os.Args, []string{schemaDirArg, keyspaceArg, numShardsArg, webDirArg, webDir2Arg, tabletHostname}...)
+	os.Args = append(os.Args, []string{schemaDirArg, keyspaceArg, numShardsArg, webDirArg, webDir2Arg, tabletHostname, vschemaDDLAuthorizedUsers}...)
 
 	cluster := runCluster()
 	defer cluster.TearDown()
 
 	assertColumnVindex(t, cluster, columnVindex{keyspace: "test_keyspace", table: "test_table", vindex: "my_vdx", vindexType: "hash", column: "id"})
 	assertColumnVindex(t, cluster, columnVindex{keyspace: "app_customer", table: "customers", vindex: "hash", vindexType: "hash", column: "id"})
+
+	// Add Hash vindex via vtgate execution on table
+	err := addColumnVindex(cluster, "test_keyspace", "alter vschema on test_table1 add vindex my_vdx (id)")
+	assert.NoError(t, err)
+	assertColumnVindex(t, cluster, columnVindex{keyspace: "test_keyspace", table: "test_table1", vindex: "my_vdx", vindexType: "hash", column: "id"})
+}
+
+func addColumnVindex(cluster vttest.LocalCluster, keyspace string, vschemaMigration string) error {
+	ctx := context.Background()
+	vtParams := mysql.ConnParams{
+		Host:   "localhost",
+		DbName: keyspace,
+		Port:   cluster.Env.PortForProtocol("vtcombo_mysql_port", ""),
+	}
+
+	conn, err := mysql.Connect(ctx, &vtParams)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+	_, err = conn.ExecuteFetch(vschemaMigration, 1, false)
+	return err
 }
 
 func assertColumnVindex(t *testing.T, cluster vttest.LocalCluster, expected columnVindex) {

go/vt/vttest/local_cluster.go

@@ -113,6 +113,9 @@ type Config struct {
 
 	// Whether to enable/disable workflow manager
 	InitWorkflowManager bool
+
+	// Authorize vschema ddl operations to a list of users
+	VSchemaDDLAuthorizedUsers string
 }
 
 // InitSchemas is a shortcut for tests that just want to setup a single

go/vt/vttest/vtprocess.go

@@ -246,6 +246,9 @@ func VtcomboProcess(env Environment, args *Config, mysql MySQLManager) *VtProces
 	if args.InitWorkflowManager {
 		vt.ExtraArgs = append(vt.ExtraArgs, []string{"-workflow_manager_init"}...)
 	}
+	if args.VSchemaDDLAuthorizedUsers != "" {
+		vt.ExtraArgs = append(vt.ExtraArgs, []string{"-vschema_ddl_authorized_users", args.VSchemaDDLAuthorizedUsers}...)
+	}
 
 	if socket != "" {
 		vt.ExtraArgs = append(vt.ExtraArgs, []string{