LookupVindex: Multiple lookup tables support for LookupVindexCreate

go/cmd/vtctldclient/command/vreplication/lookupvindex/lookupvindex.go

@@ -18,12 +18,14 @@ package lookupvindex
 
 import (
 	"fmt"
+	"os"
 	"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/json2"
 
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	vschemapb "vitess.io/vitess/go/vt/proto/vschema"
@@ -68,13 +70,38 @@ var (
 		TabletTypesInPreferenceOrder bool
 		IgnoreNulls                  bool
 		ContinueAfterCopyWithOwner   bool
+		ParamsFile                   string
 	}{}
 
 	externalizeOptions = struct {
 		Keyspace string
 	}{}
 
 	parseAndValidateCreate = func(cmd *cobra.Command, args []string) error {
+		if createOptions.ParamsFile != "" {
+			paramsFile, err := os.ReadFile(createOptions.ParamsFile)
+			if err != nil {
+				return err
+			}
+			createParams := &vtctldatapb.LookupVindexCreateParams{}
+			err = json2.UnmarshalPB(paramsFile, createParams)
+			if err != nil {
+				return err
+			}
+			return parseCreateParams(createParams)
+		}
+		if createOptions.Keyspace == "" {
+			return fmt.Errorf("keyspace is a required flag.")
+		}
+		if createOptions.TableOwner == "" {
+			return fmt.Errorf("table-owner is a required flag.")
+		}
+		if len(createOptions.TableOwnerColumns) == 0 {
+			return fmt.Errorf("table-owner-columns is a required flag.")
+		}
+		if createOptions.Keyspace == "" {
+			return fmt.Errorf("keyspace is a required flag.")
+		}
 		if createOptions.TableName == "" { // Use vindex name
 			createOptions.TableName = baseOptions.Name
 		}
@@ -130,6 +157,95 @@ var (
 		return nil
 	}
 
+	parseCreateParams = func(params *vtctldatapb.LookupVindexCreateParams) error {
+		if params.Keyspace == "" {
+			return fmt.Errorf("keyspace cannot be empty")
+		}
+		createOptions.Keyspace = params.Keyspace
+
+		if len(params.Vindexes) == 0 {
+			return fmt.Errorf("atleast 1 vindex is required")
+		}
+
+		vindexes := map[string]*vschemapb.Vindex{}
+		tables := map[string]*vschemapb.Table{}
+		for _, vindex := range params.Vindexes {
+			if vindex.TableName == "" {
+				vindex.TableName = vindex.Name
+			}
+
+			if !strings.Contains(vindex.LookupVindexType, "lookup") {
+				return fmt.Errorf("%s is not a lookup vindex type.", vindex.LookupVindexType)
+			}
+
+			vindexes[vindex.Name] = &vschemapb.Vindex{
+				Type: vindex.LookupVindexType,
+				Params: map[string]string{
+					"table":        baseOptions.TableKeyspace + "." + vindex.TableName,
+					"from":         strings.Join(vindex.TableOwnerColumns, ","),
+					"to":           "keyspace_id",
+					"ignore_nulls": fmt.Sprintf("%t", vindex.IgnoreNulls),
+				},
+				Owner: vindex.TableOwner,
+			}
+
+			targetTableColumnVindex := &vschemapb.ColumnVindex{
+				// If the vindex name/type is empty then we'll fill this in
+				// later using the defult for the column types.
+				Name:    vindex.TableVindexType,
+				Columns: vindex.TableOwnerColumns,
+			}
+			sourceTableColumnVindex := &vschemapb.ColumnVindex{
+				Name:    vindex.Name,
+				Columns: vindex.TableOwnerColumns,
+			}
+
+			if table, ok := tables[vindex.TableName]; !ok {
+				tables[vindex.TableName] = &vschemapb.Table{
+					ColumnVindexes: []*vschemapb.ColumnVindex{targetTableColumnVindex},
+				}
+			} else {
+				table.ColumnVindexes = append(table.ColumnVindexes, targetTableColumnVindex)
+			}
+
+			if table, ok := tables[vindex.TableOwner]; !ok {
+				tables[vindex.TableOwner] = &vschemapb.Table{
+					ColumnVindexes: []*vschemapb.ColumnVindex{sourceTableColumnVindex},
+				}
+			} else {
+				table.ColumnVindexes = append(table.ColumnVindexes, sourceTableColumnVindex)
+			}
+		}
+
+		baseOptions.Vschema = &vschemapb.Keyspace{
+			Vindexes: vindexes,
+			Tables:   tables,
+		}
+
+		// VReplication specific flags.
+		if len(params.TabletTypes) == 0 {
+			createOptions.TabletTypes = tabletTypesDefault
+		} else {
+			createOptions.TabletTypes = params.TabletTypes
+		}
+
+		if len(params.Cells) != 0 {
+			for i, cell := range createOptions.Cells {
+				createOptions.Cells[i] = strings.TrimSpace(cell)
+			}
+		}
+
+		if params.ContinueAfterCopyWithOwner != nil {
+			createOptions.ContinueAfterCopyWithOwner = *params.ContinueAfterCopyWithOwner
+		}
+
+		if params.TabletTypesInPreferenceOrder != nil {
+			createOptions.ContinueAfterCopyWithOwner = *params.TabletTypesInPreferenceOrder
+		}
+
+		return nil
+	}
+
 	// cancel makes a WorkflowDelete call to a vtctld.
 	cancel = &cobra.Command{
 		Use:                   "cancel",
@@ -145,7 +261,7 @@ var (
 	// create makes a LookupVindexCreate call to a vtctld.
 	create = &cobra.Command{
 		Use:                   "create",
-		Short:                 "Create the Lookup Vindex in the specified keyspace and backfill it with a VReplication workflow.",
+		Short:                 "Create the Lookup Vindex(es) in the specified keyspace and backfill them with a VReplication workflow.",
 		Example:               `vtctldclient --server localhost:15999 LookupVindex --name corder_lookup_vdx --table-keyspace customer create --keyspace customer --type consistent_lookup_unique --table-owner corder --table-owner-columns sku --table-name corder_lookup_tbl --table-vindex-type unicode_loose_xxhash`,
 		SilenceUsage:          true,
 		DisableFlagsInUseLine: true,
@@ -274,7 +390,7 @@ func commandShow(cmd *cobra.Command, args []string) error {
 }
 
 func registerCommands(root *cobra.Command) {
-	base.PersistentFlags().StringVar(&baseOptions.Name, "name", "", "The name of the Lookup Vindex to create. This will also be the name of the VReplication workflow created to backfill the Lookup Vindex.")
+	base.PersistentFlags().StringVar(&baseOptions.Name, "name", "", "The name of the Lookup Vindex to create. This will also be the name of the VReplication workflow created to backfill the Lookup Vindex. This will be used only for the workflow name if params-file is used.")
 	base.MarkPersistentFlagRequired("name")
 	base.PersistentFlags().StringVar(&baseOptions.TableKeyspace, "table-keyspace", "", "The keyspace to create the lookup table in. This is also where the VReplication workflow is created to backfill the Lookup Vindex.")
 	base.MarkPersistentFlagRequired("table-keyspace")
@@ -283,17 +399,14 @@ func registerCommands(root *cobra.Command) {
 	// This will create the lookup vindex in the specified keyspace
 	// and setup a VReplication workflow to backfill its lookup table.
 	create.Flags().StringVar(&createOptions.Keyspace, "keyspace", "", "The keyspace to create the Lookup Vindex in. This is also where the table-owner must exist.")
-	create.MarkFlagRequired("keyspace")
 	create.Flags().StringVar(&createOptions.Type, "type", "", "The type of Lookup Vindex to create.")
-	create.MarkFlagRequired("type")
 	create.Flags().StringVar(&createOptions.TableOwner, "table-owner", "", "The table holding the data which we should use to backfill the Lookup Vindex. This must exist in the same keyspace as the Lookup Vindex.")
-	create.MarkFlagRequired("table-owner")
 	create.Flags().StringSliceVar(&createOptions.TableOwnerColumns, "table-owner-columns", nil, "The columns to read from the owner table. These will be used to build the hash which gets stored as the keyspace_id value in the lookup table.")
-	create.MarkFlagRequired("table-owner-columns")
 	create.Flags().StringVar(&createOptions.TableName, "table-name", "", "The name of the lookup table. If not specified, then it will be created using the same name as the Lookup Vindex.")
 	create.Flags().StringVar(&createOptions.TableVindexType, "table-vindex-type", "", "The primary vindex name/type to use for the lookup table, if the table-keyspace is sharded. If no value is provided then the default type will be used based on the table-owner-columns types.")
 	create.Flags().BoolVar(&createOptions.IgnoreNulls, "ignore-nulls", false, "Do not add corresponding records in the lookup table if any of the owner table's 'from' fields are NULL.")
 	create.Flags().BoolVar(&createOptions.ContinueAfterCopyWithOwner, "continue-after-copy-with-owner", true, "Vindex will continue materialization after the backfill completes when an owner is provided.")
+	create.Flags().StringVar(&createOptions.ParamsFile, "params-file", "", "JSON file containing lookup vindex create options. Use this for creating multiple lookup vindexes.")
 	// VReplication specific flags.
 	create.Flags().StringSliceVar(&createOptions.Cells, "cells", nil, "Cells to look in for source tablets to replicate from.")
 	create.Flags().Var((*topoprotopb.TabletTypeListFlag)(&createOptions.TabletTypes), "tablet-types", "Source tablet types to replicate from.")