Migrate Materialize command to vtctldclient

go/cmd/vtctldclient/command/root.go

@@ -29,6 +29,7 @@ import (
 	// These imports ensure init()s within them get called and they register their commands/subcommands.
 	vreplcommon "vitess.io/vitess/go/cmd/vtctldclient/command/vreplication/common"
 	_ "vitess.io/vitess/go/cmd/vtctldclient/command/vreplication/lookupvindex"
+	_ "vitess.io/vitess/go/cmd/vtctldclient/command/vreplication/materialize"
 	_ "vitess.io/vitess/go/cmd/vtctldclient/command/vreplication/movetables"
 	_ "vitess.io/vitess/go/cmd/vtctldclient/command/vreplication/reshard"
 	_ "vitess.io/vitess/go/cmd/vtctldclient/command/vreplication/vdiff"

go/cmd/vtctldclient/command/vreplication/common/cancel.go

@@ -27,7 +27,7 @@ import (
 	vtctldatapb "vitess.io/vitess/go/vt/proto/vtctldata"
 )
 
-var cancelOptions = struct {
+var CancelOptions = struct {
 	KeepData         bool
 	KeepRoutingRules bool
 }{}
@@ -56,8 +56,8 @@ func commandCancel(cmd *cobra.Command, args []string) error {
 	req := &vtctldatapb.WorkflowDeleteRequest{
 		Keyspace:         BaseOptions.TargetKeyspace,
 		Workflow:         BaseOptions.Workflow,
-		KeepData:         cancelOptions.KeepData,
-		KeepRoutingRules: cancelOptions.KeepRoutingRules,
+		KeepData:         CancelOptions.KeepData,
+		KeepRoutingRules: CancelOptions.KeepRoutingRules,
 	}
 	resp, err := GetClient().WorkflowDelete(GetCommandCtx(), req)
 	if err != nil {

go/cmd/vtctldclient/command/vreplication/common/update.go

@@ -127,14 +127,14 @@ func commandUpdateState(cmd *cobra.Command, args []string) error {
 	var state binlogdatapb.VReplicationWorkflowState
 	switch strings.ToLower(cmd.Name()) {
 	case "start":
-		if err := CanRestartWorkflow(workflowUpdateOptions.Workflow, workflowOptions.Keyspace); err != nil {
+		if err := CanRestartWorkflow(workflowOptions.Keyspace, workflowUpdateOptions.Workflow); err != nil {
 			return err
 		}
 		state = binlogdatapb.VReplicationWorkflowState_Running
 	case "stop":
 		state = binlogdatapb.VReplicationWorkflowState_Stopped
 	default:
-		return fmt.Errorf("invalid workstate: %s", args[0])
+		return fmt.Errorf("invalid workflow state: %s", args[0])
 	}
 
 	// The only thing we're updating is the state.

go/cmd/vtctldclient/command/vreplication/common/utils.go

@@ -185,8 +185,8 @@ func AddCommonCreateFlags(cmd *cobra.Command) {
 	cmd.Flags().BoolVar(&CreateOptions.TabletTypesInPreferenceOrder, "tablet-types-in-preference-order", true, "When performing source tablet selection, look for candidates in the type order as they are listed in the tablet-types flag.")
 	cmd.Flags().StringVar(&CreateOptions.OnDDL, "on-ddl", onDDLDefault, "What to do when DDL is encountered in the VReplication stream. Possible values are IGNORE, STOP, EXEC, and EXEC_IGNORE.")
 	cmd.Flags().BoolVar(&CreateOptions.DeferSecondaryKeys, "defer-secondary-keys", false, "Defer secondary index creation for a table until after it has been copied.")
-	cmd.Flags().BoolVar(&CreateOptions.AutoStart, "auto-start", true, "Start the MoveTables workflow after creating it.")
-	cmd.Flags().BoolVar(&CreateOptions.StopAfterCopy, "stop-after-copy", false, "Stop the MoveTables workflow after it's finished copying the existing rows and before it starts replicating changes.")
+	cmd.Flags().BoolVar(&CreateOptions.AutoStart, "auto-start", true, "Start the workflow after creating it.")
+	cmd.Flags().BoolVar(&CreateOptions.StopAfterCopy, "stop-after-copy", false, "Stop the workflow after it's finished copying the existing rows and before it starts replicating changes.")
 }
 
 var SwitchTrafficOptions = struct {

go/cmd/vtctldclient/command/vreplication/materialize/create.go

@@ -0,0 +1,162 @@
+/*
+Copyright 2023 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package materialize
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/spf13/cobra"
+
+	"vitess.io/vitess/go/cmd/vtctldclient/cli"
+	"vitess.io/vitess/go/cmd/vtctldclient/command/vreplication/common"
+	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/topo/topoproto"
+
+	vtctldatapb "vitess.io/vitess/go/vt/proto/vtctldata"
+)
+
+var (
+	createOptions = struct {
+		SourceKeyspace string
+		TableSettings  tableSettings
+	}{}
+
+	// create makes a MaterializeCreate gRPC call to a vtctld.
+	create = &cobra.Command{
+		Use:                   "create",
+		Short:                 "Create and run a Materialize VReplication workflow.",
+		Example:               `vtctldclient --server localhost:15999 materialize --workflow product_sales --target-keyspace commerce create --source-keyspace commerce --table-settings '[{"target_table": "sales_by_sku", "create_ddl": "create table sales_by_sku (sku varbinary(128) not null primary key, orders bigint, revenue bigint)", "source_expression": "select sku, count(*) as orders, sum(price) as revenue from corder group by sku"}]' --cells zone1 --cells zone2 --tablet-types replicas`,
+		SilenceUsage:          true,
+		DisableFlagsInUseLine: true,
+		Aliases:               []string{"Create"},
+		Args:                  cobra.NoArgs,
+		PreRunE: func(cmd *cobra.Command, args []string) error {
+			if err := common.ParseAndValidateCreateOptions(cmd); err != nil {
+				return err
+			}
+			return nil
+		},
+		RunE: commandCreate,
+	}
+)
+
+func commandCreate(cmd *cobra.Command, args []string) error {
+	format, err := common.GetOutputFormat(cmd)
+	if err != nil {
+		return err
+	}
+	tsp := common.GetTabletSelectionPreference(cmd)
+	cli.FinishedParsing(cmd)
+
+	ms := &vtctldatapb.MaterializeSettings{
+		Workflow:                  common.BaseOptions.Workflow,
+		TargetKeyspace:            common.BaseOptions.TargetKeyspace,
+		SourceKeyspace:            createOptions.SourceKeyspace,
+		TableSettings:             createOptions.TableSettings.val,
+		StopAfterCopy:             common.CreateOptions.StopAfterCopy,
+		Cell:                      strings.Join(common.CreateOptions.Cells, ","),
+		TabletTypes:               topoproto.MakeStringTypeCSV(common.CreateOptions.TabletTypes),
+		TabletSelectionPreference: tsp,
+	}
+
+	req := &vtctldatapb.MaterializeCreateRequest{
+		Settings: ms,
+	}
+
+	_, err = common.GetClient().MaterializeCreate(common.GetCommandCtx(), req)
+	if err != nil {
+		return err
+	}
+
+	if format == "json" {
+		resp := struct {
+			Action string
+			Status string
+		}{
+			Action: "create",
+			Status: "success",
+		}
+		jsonText, _ := cli.MarshalJSONPretty(resp)
+		fmt.Println(string(jsonText))
+	} else {
+		fmt.Printf("Materialization workflow %s successfully created in the %s keyspace. Use show to view the status.\n",
+			common.BaseOptions.Workflow, common.BaseOptions.TargetKeyspace)
+	}
+
+	return nil
+}
+
+// tableSettings is a wrapper around a slice of TableMaterializeSettings
+// proto messages that implements the pflag.Value interface.
+type tableSettings struct {
+	val []*vtctldatapb.TableMaterializeSettings
+}
+
+func (ts *tableSettings) String() string {
+	tsj, _ := json.Marshal(ts.val)
+	return string(tsj)
+}
+
+func (ts *tableSettings) Set(v string) error {
+	ts.val = make([]*vtctldatapb.TableMaterializeSettings, 0)
+	err := json.Unmarshal([]byte(v), &ts.val)
+	if err != nil {
+		return fmt.Errorf("table-settings is not valid JSON")
+	}
+	if len(ts.val) == 0 {
+		return fmt.Errorf("empty table-settings")
+	}
+
+	// Validate the provided queries.
+	seenSourceTables := make(map[string]bool)
+	for _, tms := range ts.val {
+		if tms.TargetTable == "" || tms.SourceExpression == "" {
+			return fmt.Errorf("missing target_table or source_expression")
+		}
+		// Validate that the query is valid.
+		stmt, err := sqlparser.Parse(tms.SourceExpression)
+		if err != nil {
+			return fmt.Errorf("invalid source_expression: %q", tms.SourceExpression)
+		}
+		// Validate that each source-expression uses a different table.
+		// If any of them query the same table the materialize workflow
+		// will fail.
+		err = sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
+			switch node := node.(type) {
+			case sqlparser.TableName:
+				if !node.Name.IsEmpty() {
+					if seenSourceTables[node.Name.String()] {
+						return false, fmt.Errorf("multiple source_expression queries use the same table: %q", node.Name.String())
+					}
+					seenSourceTables[node.Name.String()] = true
+				}
+			}
+			return true, nil
+		}, stmt)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (ts *tableSettings) Type() string {
+	return "JSON"
+}

go/cmd/vtctldclient/command/vreplication/materialize/materialize.go

@@ -0,0 +1,64 @@
+/*
+Copyright 2023 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package materialize
+
+import (
+	"github.com/spf13/cobra"
+
+	"vitess.io/vitess/go/cmd/vtctldclient/command/vreplication/common"
+	"vitess.io/vitess/go/vt/topo/topoproto"
+)
+
+var (
+	// base is the base command for all actions related to Materialize.
+	base = &cobra.Command{
+		Use:                   "Materialize --workflow <workflow> --target-keyspace <keyspace> [command] [command-flags]",
+		Short:                 "Perform commands related to materializing query results from the source keyspace into tables in the target keyspace.",
+		DisableFlagsInUseLine: true,
+		Aliases:               []string{"materialize"},
+		Args:                  cobra.ExactArgs(1),
+	}
+)
+
+func registerCommands(root *cobra.Command) {
+	common.AddCommonFlags(base)
+	root.AddCommand(base)
+
+	create.Flags().StringSliceVarP(&common.CreateOptions.Cells, "cells", "c", nil, "Cells and/or CellAliases to copy table data from.")
+	create.Flags().Var((*topoproto.TabletTypeListFlag)(&common.CreateOptions.TabletTypes), "tablet-types", "Source tablet types to replicate table data from (e.g. PRIMARY,REPLICA,RDONLY).")
+	create.Flags().BoolVar(&common.CreateOptions.TabletTypesInPreferenceOrder, "tablet-types-in-preference-order", true, "When performing source tablet selection, look for candidates in the type order as they are listed in the tablet-types flag.")
+	create.Flags().StringVar(&createOptions.SourceKeyspace, "source-keyspace", "", "Keyspace where the tables queried in the 'source_expression' values within table-settings live.")
+	create.MarkFlagRequired("source-keyspace")
+	create.Flags().Var(&createOptions.TableSettings, "table-settings", "A JSON array where each value must contain two key/value pairs. The first required key is 'target_table' and it is the name of the table in the target-keyspace to store the results in. The second required key is 'source_expression' and its value is the select query to run against the source table. An optional k/v pair can also be specified for 'create_ddl' which provides the DDL to create the target table if it does not exist (you can specify a value of 'copy' if the target-table should be copied as-is from the source keyspace).")
+	create.MarkFlagRequired("table-settings")
+	create.Flags().BoolVar(&common.CreateOptions.StopAfterCopy, "stop-after-copy", false, "Stop the workflow after it's finished copying the existing rows and before it starts replicating changes.")
+	base.AddCommand(create)
+
+	// Generic workflow commands.
+	opts := &common.SubCommandsOpts{
+		SubCommand: "Materialize",
+		Workflow:   "product_sales",
+	}
+	base.AddCommand(common.GetCancelCommand(opts))
+	base.AddCommand(common.GetShowCommand(opts))
+	base.AddCommand(common.GetStartCommand(opts))
+	base.AddCommand(common.GetStopCommand(opts))
+}
+
+func init() {
+	common.RegisterCommandHandler("Materialize", registerCommands)
+}

go/cmd/vtctldclient/command/vreplication/movetables/create.go

@@ -29,7 +29,7 @@ import (
 )
 
 var (
-	moveTablesCreateOptions = struct {
+	createOptions = struct {
 		SourceKeyspace      string
 		SourceShards        []string
 		ExternalClusterName string
@@ -41,8 +41,8 @@ var (
 		AtomicCopy          bool
 	}{}
 
-	// moveTablesCreate makes a moveTablesCreate gRPC call to a vtctld.
-	moveTablesCreate = &cobra.Command{
+	// create makes a MoveTablesCreate gRPC call to a vtctld.
+	create = &cobra.Command{
 		Use:                   "create",
 		Short:                 "Create and optionally run a MoveTables VReplication workflow.",
 		Example:               `vtctldclient --server localhost:15999 movetables --workflow commerce2customer --target-keyspace customer create --source-keyspace commerce --cells zone1 --cells zone2 --tablet-types replica`,
@@ -60,13 +60,13 @@ var (
 			}
 			checkAtomicCopyOptions := func() error {
 				var errors []string
-				if !moveTablesCreateOptions.AtomicCopy {
+				if !createOptions.AtomicCopy {
 					return nil
 				}
-				if !moveTablesCreateOptions.AllTables {
+				if !createOptions.AllTables {
 					errors = append(errors, "atomic copy requires --all-tables")
 				}
-				if len(moveTablesCreateOptions.IncludeTables) > 0 || len(moveTablesCreateOptions.ExcludeTables) > 0 {
+				if len(createOptions.IncludeTables) > 0 || len(createOptions.ExcludeTables) > 0 {
 					errors = append(errors, "atomic copy does not support specifying tables")
 				}
 				if len(errors) > 0 {
@@ -79,11 +79,11 @@ var (
 			}
 			return nil
 		},
-		RunE: commandMoveTablesCreate,
+		RunE: commandCreate,
 	}
 )
 
-func commandMoveTablesCreate(cmd *cobra.Command, args []string) error {
+func commandCreate(cmd *cobra.Command, args []string) error {
 	format, err := common.GetOutputFormat(cmd)
 	if err != nil {
 		return err
@@ -94,21 +94,21 @@ func commandMoveTablesCreate(cmd *cobra.Command, args []string) error {
 	req := &vtctldatapb.MoveTablesCreateRequest{
 		Workflow:                  common.BaseOptions.Workflow,
 		TargetKeyspace:            common.BaseOptions.TargetKeyspace,
-		SourceKeyspace:            moveTablesCreateOptions.SourceKeyspace,
-		SourceShards:              moveTablesCreateOptions.SourceShards,
-		SourceTimeZone:            moveTablesCreateOptions.SourceTimeZone,
+		SourceKeyspace:            createOptions.SourceKeyspace,
+		SourceShards:              createOptions.SourceShards,
+		SourceTimeZone:            createOptions.SourceTimeZone,
 		Cells:                     common.CreateOptions.Cells,
 		TabletTypes:               common.CreateOptions.TabletTypes,
 		TabletSelectionPreference: tsp,
-		AllTables:                 moveTablesCreateOptions.AllTables,
-		IncludeTables:             moveTablesCreateOptions.IncludeTables,
-		ExcludeTables:             moveTablesCreateOptions.ExcludeTables,
+		AllTables:                 createOptions.AllTables,
+		IncludeTables:             createOptions.IncludeTables,
+		ExcludeTables:             createOptions.ExcludeTables,
 		OnDdl:                     common.CreateOptions.OnDDL,
 		DeferSecondaryKeys:        common.CreateOptions.DeferSecondaryKeys,
 		AutoStart:                 common.CreateOptions.AutoStart,
 		StopAfterCopy:             common.CreateOptions.StopAfterCopy,
-		NoRoutingRules:            moveTablesCreateOptions.NoRoutingRules,
-		AtomicCopy:                moveTablesCreateOptions.AtomicCopy,
+		NoRoutingRules:            createOptions.NoRoutingRules,
+		AtomicCopy:                createOptions.AtomicCopy,
 	}
 
 	resp, err := common.GetClient().MoveTablesCreate(common.GetCommandCtx(), req)
@@ -120,17 +120,3 @@ func commandMoveTablesCreate(cmd *cobra.Command, args []string) error {
 	}
 	return nil
 }
-
-func registerCreateCommand(root *cobra.Command) {
-	common.AddCommonCreateFlags(moveTablesCreate)
-	moveTablesCreate.PersistentFlags().StringVar(&moveTablesCreateOptions.SourceKeyspace, "source-keyspace", "", "Keyspace where the tables are being moved from.")
-	moveTablesCreate.MarkPersistentFlagRequired("source-keyspace")
-	moveTablesCreate.Flags().StringSliceVar(&moveTablesCreateOptions.SourceShards, "source-shards", nil, "Source shards to copy data from when performing a partial moveTables (experimental).")
-	moveTablesCreate.Flags().StringVar(&moveTablesCreateOptions.SourceTimeZone, "source-time-zone", "", "Specifying this causes any DATETIME fields to be converted from the given time zone into UTC.")
-	moveTablesCreate.Flags().BoolVar(&moveTablesCreateOptions.AllTables, "all-tables", false, "Copy all tables from the source.")
-	moveTablesCreate.Flags().StringSliceVar(&moveTablesCreateOptions.IncludeTables, "tables", nil, "Source tables to copy.")
-	moveTablesCreate.Flags().StringSliceVar(&moveTablesCreateOptions.ExcludeTables, "exclude-tables", nil, "Source tables to exclude from copying.")
-	moveTablesCreate.Flags().BoolVar(&moveTablesCreateOptions.NoRoutingRules, "no-routing-rules", false, "(Advanced) Do not create routing rules while creating the workflow. See the reference documentation for limitations if you use this flag.")
-	moveTablesCreate.Flags().BoolVar(&moveTablesCreateOptions.AtomicCopy, "atomic-copy", false, "(EXPERIMENTAL) A single copy phase is run for all tables from the source. Use this, for example, if your source keyspace has tables which use foreign key constraints.")
-	moveTables.AddCommand(moveTablesCreate)
-}