Add pscale branch vtctld move-tables start

AGENTS.md

@@ -479,14 +479,18 @@ External create required flags: `--host`, `--source-database`, `--username`, `--
 
 ## Vitess MoveTables
 
-Copy tables between keyspaces with `pscale branch vtctld move-tables`. `pscale workflow` will be deprecated soon; prefer `move-tables` for new work. JSON output includes `next_steps` — follow those commands. Typical order: create the target keyspace (`keyspace create` or `keyspace create-external`), create the workflow, poll `status`, switch replica traffic, then primary traffic (ask the user first), then `complete --dry-run` and `complete` after approval.
+Copy tables between keyspaces with `pscale branch vtctld move-tables`. `pscale workflow` will be deprecated soon; prefer `move-tables` for new work. JSON output includes `next_steps` — follow those commands. Typical order: create the target keyspace (`keyspace create` or `keyspace create-external`), create the workflow, poll `status`, switch replica traffic, then primary traffic (ask the user first), then `complete --dry-run` and `complete` after approval. Create with `--auto-start=false` to leave streams stopped, then `start` when you are ready.
 
 `--workflow` is the workflow name you choose. `--source-keyspace` and `--target-keyspace` are required on create. Pass `--tables t1,t2` or `--all-tables` (mutually exclusive).
 
 ```bash
 pscale branch vtctld move-tables list <database> <branch> --org <org> --format json
 pscale branch vtctld move-tables create <database> <branch> --org <org> --format json \
   --workflow <workflow> --source-keyspace <source> --target-keyspace <target> --tables <table>
+pscale branch vtctld move-tables create <database> <branch> --org <org> --format json \
+  --workflow <workflow> --source-keyspace <source> --target-keyspace <target> --tables <table> --auto-start=false
+pscale branch vtctld move-tables start <database> <branch> --org <org> --format json \
+  --workflow <workflow> --target-keyspace <target>
 pscale branch vtctld move-tables status <database> <branch> --org <org> --format json \
   --workflow <workflow> --target-keyspace <target>
 pscale branch vtctld move-tables switch-traffic <database> <branch> --org <org> --format json \

internal/cmd/branch/vtctld/move_tables.go

@@ -29,6 +29,7 @@ func MoveTablesCmd(ch *cmdutil.Helper) *cobra.Command {
 	cmd.AddCommand(MoveTablesCreateCmd(ch))
 	cmd.AddCommand(MoveTablesShowCmd(ch))
 	cmd.AddCommand(MoveTablesStatusCmd(ch))
+	cmd.AddCommand(MoveTablesStartCmd(ch))
 	cmd.AddCommand(MoveTablesSwitchTrafficCmd(ch))
 	cmd.AddCommand(MoveTablesReverseTrafficCmd(ch))
 	cmd.AddCommand(MoveTablesCancelCmd(ch))
@@ -300,6 +301,56 @@ func MoveTablesStatusCmd(ch *cmdutil.Helper) *cobra.Command {
 	return cmd
 }
 
+func MoveTablesStartCmd(ch *cmdutil.Helper) *cobra.Command {
+	var flags struct {
+		workflow       string
+		targetKeyspace string
+	}
+
+	cmd := &cobra.Command{
+		Use:   "start <database> <branch>",
+		Short: "Start a MoveTables workflow",
+		Args:  cmdutil.RequiredArgs("database", "branch"),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx := cmd.Context()
+			database, branch := args[0], args[1]
+
+			client, err := ch.Client()
+			if err != nil {
+				return err
+			}
+
+			end := ch.Printer.PrintProgress(
+				fmt.Sprintf("Starting MoveTables workflow %s on %s\u2026",
+					printer.BoldBlue(flags.workflow), progressTarget(ch.Config.Organization, database, branch)))
+			defer end()
+
+			data, err := client.MoveTables.Start(ctx, &ps.MoveTablesStartRequest{
+				Organization:   ch.Config.Organization,
+				Database:       database,
+				Branch:         branch,
+				Workflow:       flags.workflow,
+				TargetKeyspace: flags.targetKeyspace,
+			})
+			if err != nil {
+				return cmdutil.HandleError(err)
+			}
+
+			end()
+			return printWorkflowJSON(ch.Printer, data, []workflowNextStep{
+				moveTablesStatusStep(ch.Config.Organization, database, branch, flags.workflow, flags.targetKeyspace, "Monitor copy and replication progress"),
+			})
+		},
+	}
+
+	cmd.Flags().StringVar(&flags.workflow, "workflow", "", "Name of the workflow")
+	cmd.Flags().StringVar(&flags.targetKeyspace, "target-keyspace", "", "Target keyspace")
+	cmd.MarkFlagRequired("workflow")        // nolint:errcheck
+	cmd.MarkFlagRequired("target-keyspace") // nolint:errcheck
+
+	return cmd
+}
+
 func MoveTablesSwitchTrafficCmd(ch *cmdutil.Helper) *cobra.Command {
 	var flags struct {
 		workflow                  string

internal/cmd/branch/vtctld/move_tables_test.go

@@ -717,6 +717,44 @@ func TestMoveTablesShow(t *testing.T) {
 	c.Assert(svc.ShowFnInvoked, qt.IsTrue)
 }
 
+func TestMoveTablesStart(t *testing.T) {
+	c := qt.New(t)
+
+	org := "my-org"
+	db := "my-db"
+	branch := "my-branch"
+
+	svc := &mock.MoveTablesService{
+		StartFn: func(ctx context.Context, req *ps.MoveTablesStartRequest) (json.RawMessage, error) {
+			c.Assert(req.Organization, qt.Equals, org)
+			c.Assert(req.Database, qt.Equals, db)
+			c.Assert(req.Branch, qt.Equals, branch)
+			c.Assert(req.Workflow, qt.Equals, "my-workflow")
+			c.Assert(req.TargetKeyspace, qt.Equals, "target-ks")
+			return json.RawMessage(`{"summary":"Streams started"}`), nil
+		},
+	}
+
+	var buf bytes.Buffer
+	ch := moveTablesTestHelper(org, svc, nil, &buf)
+
+	cmd := MoveTablesCmd(ch)
+	cmd.SetArgs([]string{"start", db, branch,
+		"--workflow", "my-workflow",
+		"--target-keyspace", "target-ks",
+	})
+	err := cmd.Execute()
+	c.Assert(err, qt.IsNil)
+	c.Assert(svc.StartFnInvoked, qt.IsTrue)
+	c.Assert(buf.String(), qt.JSONEquals, map[string]any{
+		"summary": "Streams started",
+		"next_steps": []any{map[string]any{
+			"command": "pscale branch vtctld move-tables status my-db my-branch --org my-org --workflow my-workflow --target-keyspace target-ks --format json",
+			"reason":  "Monitor copy and replication progress",
+		}},
+	})
+}
+
 func TestMoveTablesStatusAddsNextSteps(t *testing.T) {
 	c := qt.New(t)
 

internal/mock/vtctld_move_tables.go

@@ -20,6 +20,9 @@ type MoveTablesService struct {
 	StatusFn        func(context.Context, *ps.MoveTablesStatusRequest) (json.RawMessage, error)
 	StatusFnInvoked bool
 
+	StartFn        func(context.Context, *ps.MoveTablesStartRequest) (json.RawMessage, error)
+	StartFnInvoked bool
+
 	SwitchTrafficFn        func(context.Context, *ps.MoveTablesSwitchTrafficRequest) (*ps.VtctldOperationReference, error)
 	SwitchTrafficFnInvoked bool
 
@@ -53,6 +56,11 @@ func (s *MoveTablesService) Status(ctx context.Context, req *ps.MoveTablesStatus
 	return s.StatusFn(ctx, req)
 }
 
+func (s *MoveTablesService) Start(ctx context.Context, req *ps.MoveTablesStartRequest) (json.RawMessage, error) {
+	s.StartFnInvoked = true
+	return s.StartFn(ctx, req)
+}
+
 func (s *MoveTablesService) SwitchTraffic(ctx context.Context, req *ps.MoveTablesSwitchTrafficRequest) (*ps.VtctldOperationReference, error) {
 	s.SwitchTrafficFnInvoked = true
 	return s.SwitchTrafficFn(ctx, req)

internal/planetscale/vtctld_move_tables.go

@@ -16,6 +16,7 @@ type MoveTablesService interface {
 	Create(context.Context, *MoveTablesCreateRequest) (*VtctldOperationReference, error)
 	Show(context.Context, *MoveTablesShowRequest) (json.RawMessage, error)
 	Status(context.Context, *MoveTablesStatusRequest) (json.RawMessage, error)
+	Start(context.Context, *MoveTablesStartRequest) (json.RawMessage, error)
 	SwitchTraffic(context.Context, *MoveTablesSwitchTrafficRequest) (*VtctldOperationReference, error)
 	ReverseTraffic(context.Context, *MoveTablesReverseTrafficRequest) (*VtctldOperationReference, error)
 	Cancel(context.Context, *MoveTablesCancelRequest) (*VtctldOperationReference, error)
@@ -73,6 +74,15 @@ type MoveTablesStatusRequest struct {
 	TargetKeyspace string `json:"-"`
 }
 
+// MoveTablesStartRequest is a request for starting a MoveTables workflow.
+type MoveTablesStartRequest struct {
+	Organization   string `json:"-"`
+	Database       string `json:"-"`
+	Branch         string `json:"-"`
+	Workflow       string `json:"-"`
+	TargetKeyspace string `json:"target_keyspace"`
+}
+
 // MoveTablesSwitchTrafficRequest is a request for switching traffic for a MoveTables workflow.
 type MoveTablesSwitchTrafficRequest struct {
 	Organization              string   `json:"-"`
@@ -192,6 +202,19 @@ func (s *moveTablesService) Status(ctx context.Context, req *MoveTablesStatusReq
 	return resp.Data, nil
 }
 
+func (s *moveTablesService) Start(ctx context.Context, req *MoveTablesStartRequest) (json.RawMessage, error) {
+	p := path.Join(moveTablesWorkflowAPIPath(req.Organization, req.Database, req.Branch, req.Workflow), "start")
+	httpReq, err := s.client.newRequest(http.MethodPost, p, req)
+	if err != nil {
+		return nil, fmt.Errorf("error creating http request: %w", err)
+	}
+	resp := &vtctldDataResponse{}
+	if err := s.client.do(ctx, httpReq, resp); err != nil {
+		return nil, err
+	}
+	return resp.Data, nil
+}
+
 func (s *moveTablesService) SwitchTraffic(ctx context.Context, req *MoveTablesSwitchTrafficRequest) (*VtctldOperationReference, error) {
 	p := path.Join(moveTablesWorkflowAPIPath(req.Organization, req.Database, req.Branch, req.Workflow), "switch-traffic")
 	return s.enqueueOperation(ctx, p, req)

internal/planetscale/vtctld_move_tables_test.go

@@ -321,6 +321,39 @@ func TestMoveTables_Status(t *testing.T) {
 	c.Assert(string(data), qt.Equals, `{"result":"ok"}`)
 }
 
+func TestMoveTables_Start(t *testing.T) {
+	c := qt.New(t)
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		c.Assert(r.Method, qt.Equals, http.MethodPost)
+		c.Assert(r.URL.Path, qt.Equals, "/v1/organizations/my-org/databases/my-db/branches/my-branch/move-tables/workflows/my-workflow/start")
+
+		var body map[string]interface{}
+		err := json.NewDecoder(r.Body).Decode(&body)
+		c.Assert(err, qt.IsNil)
+		c.Assert(body["target_keyspace"], qt.Equals, "target")
+
+		w.WriteHeader(http.StatusOK)
+		_, err = w.Write([]byte(`{"data":{"summary":"Streams started"}}`))
+		c.Assert(err, qt.IsNil)
+	}))
+	defer ts.Close()
+
+	client, err := NewClient(WithBaseURL(ts.URL))
+	c.Assert(err, qt.IsNil)
+
+	ctx := context.Background()
+	data, err := client.MoveTables.Start(ctx, &MoveTablesStartRequest{
+		Organization:   "my-org",
+		Database:       "my-db",
+		Branch:         "my-branch",
+		Workflow:       "my-workflow",
+		TargetKeyspace: "target",
+	})
+	c.Assert(err, qt.IsNil)
+	c.Assert(string(data), qt.Equals, `{"summary":"Streams started"}`)
+}
+
 func TestMoveTables_SwitchTraffic(t *testing.T) {
 	c := qt.New(t)