LookupVindex: Implement internalize command for lookup vindexes

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

@@ -74,6 +74,14 @@ var (
 		Keyspace string
 	}{}
 
+	internalizeOptions = struct {
+		Keyspace string
+	}{}
+
+	completeOptions = struct {
+		Keyspace string
+	}{}
+
 	parseAndValidateCreate = func(cmd *cobra.Command, args []string) error {
 		if createOptions.TableName == "" { // Use vindex name
 			createOptions.TableName = baseOptions.Name
@@ -142,6 +150,18 @@ var (
 		RunE:                  commandCancel,
 	}
 
+	// complete makes a LookupVindexComplete call to a vtctld.
+	complete = &cobra.Command{
+		Use:                   "complete",
+		Short:                 "Complete the Lookup Vindex. If the Vindex has an owner the VReplication workflow will also be deleted.",
+		Example:               `vtctldclient --server localhost:15999 LookupVindex --name corder_lookup_vdx --table-keyspace customer complete`,
+		SilenceUsage:          true,
+		DisableFlagsInUseLine: true,
+		Aliases:               []string{"Complete"},
+		Args:                  cobra.NoArgs,
+		RunE:                  commandComplete,
+	}
+
 	// create makes a LookupVindexCreate call to a vtctld.
 	create = &cobra.Command{
 		Use:                   "create",
@@ -158,7 +178,7 @@ var (
 	// externalize makes a LookupVindexExternalize call to a vtctld.
 	externalize = &cobra.Command{
 		Use:                   "externalize",
-		Short:                 "Externalize the Lookup Vindex. If the Vindex has an owner the VReplication workflow will also be deleted.",
+		Short:                 "Externalize the Lookup Vindex. If the Vindex has an owner the VReplication workflow will also be stopped.",
 		Example:               `vtctldclient --server localhost:15999 LookupVindex --name corder_lookup_vdx --table-keyspace customer externalize`,
 		SilenceUsage:          true,
 		DisableFlagsInUseLine: true,
@@ -167,6 +187,18 @@ var (
 		RunE:                  commandExternalize,
 	}
 
+	// internalize makes a LookupVindexInternalize call to a vtctld.
+	internalize = &cobra.Command{
+		Use:                   "internalize",
+		Short:                 "Internalize the Lookup Vindex. If the Vindex has an owner the VReplication workflow will also be started.",
+		Example:               `vtctldclient --server localhost:15999 LookupVindex --name corder_lookup_vdx --table-keyspace customer internalize`,
+		SilenceUsage:          true,
+		DisableFlagsInUseLine: true,
+		Aliases:               []string{"Internalize"},
+		Args:                  cobra.NoArgs,
+		RunE:                  commandInternalize,
+	}
+
 	// show makes a GetWorkflows call to a vtctld.
 	show = &cobra.Command{
 		Use:                   "show",
@@ -199,6 +231,33 @@ func commandCancel(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
+func commandComplete(cmd *cobra.Command, args []string) error {
+	if completeOptions.Keyspace == "" {
+		completeOptions.Keyspace = baseOptions.TableKeyspace
+	}
+	cli.FinishedParsing(cmd)
+
+	resp, err := common.GetClient().LookupVindexComplete(common.GetCommandCtx(), &vtctldatapb.LookupVindexCompleteRequest{
+		Keyspace: completeOptions.Keyspace,
+		// The name of the workflow and lookup vindex.
+		Name: baseOptions.Name,
+		// Where the lookup table and VReplication workflow were created.
+		TableKeyspace: baseOptions.TableKeyspace,
+	})
+
+	if err != nil {
+		return err
+	}
+
+	output := fmt.Sprintf("LookupVindex %s has been completed", baseOptions.Name)
+	if resp.WorkflowDeleted {
+		output = output + fmt.Sprintf(" and the %s VReplication workflow has been deleted", baseOptions.Name)
+	}
+	fmt.Println(output)
+
+	return nil
+}
+
 func commandCreate(cmd *cobra.Command, args []string) error {
 	tsp := common.GetTabletSelectionPreference(cmd)
 	cli.FinishedParsing(cmd)
@@ -243,8 +302,35 @@ func commandExternalize(cmd *cobra.Command, args []string) error {
 	}
 
 	output := fmt.Sprintf("LookupVindex %s has been externalized", baseOptions.Name)
-	if resp.WorkflowDeleted {
-		output = output + fmt.Sprintf(" and the %s VReplication workflow has been deleted", baseOptions.Name)
+	if resp.WorkflowStopped {
+		output = output + fmt.Sprintf(" and the %s VReplication workflow has been stopped", baseOptions.Name)
+	}
+	fmt.Println(output)
+
+	return nil
+}
+
+func commandInternalize(cmd *cobra.Command, args []string) error {
+	if internalizeOptions.Keyspace == "" {
+		internalizeOptions.Keyspace = baseOptions.TableKeyspace
+	}
+	cli.FinishedParsing(cmd)
+
+	resp, err := common.GetClient().LookupVindexInternalize(common.GetCommandCtx(), &vtctldatapb.LookupVindexInternalizeRequest{
+		Keyspace: internalizeOptions.Keyspace,
+		// The name of the workflow and lookup vindex.
+		Name: baseOptions.Name,
+		// Where the lookup table and VReplication workflow were created.
+		TableKeyspace: baseOptions.TableKeyspace,
+	})
+
+	if err != nil {
+		return err
+	}
+
+	output := fmt.Sprintf("LookupVindex %s has been internalized", baseOptions.Name)
+	if resp.WorkflowStarted {
+		output = output + fmt.Sprintf(" and the %s VReplication workflow has been started", baseOptions.Name)
 	}
 	fmt.Println(output)
 
@@ -304,12 +390,18 @@ func registerCommands(root *cobra.Command) {
 	// for the VReplication workflow used.
 	base.AddCommand(show)
 
-	// This will also delete the VReplication workflow if the
+	// This will also stop the VReplication workflow if the
 	// vindex has an owner as the lookup vindex will then be
 	// managed by VTGate.
 	externalize.Flags().StringVar(&externalizeOptions.Keyspace, "keyspace", "", "The keyspace containing the Lookup Vindex. If no value is specified then the table-keyspace will be used.")
 	base.AddCommand(externalize)
 
+	internalize.Flags().StringVar(&internalizeOptions.Keyspace, "keyspace", "", "The keyspace containing the Lookup Vindex. If no value is specified then the table-keyspace will be used.")
+	base.AddCommand(internalize)
+
+	complete.Flags().StringVar(&completeOptions.Keyspace, "keyspace", "", "The keyspace containing the Lookup Vindex. If no value is specified then the table-keyspace will be used.")
+	base.AddCommand(complete)
+
 	// The cancel command deletes the VReplication workflow used
 	// to backfill the lookup vindex. It ends up making a
 	// WorkflowDelete VtctldServer call.

go/test/endtoend/vreplication/lookup_vindex_helper_test.go

@@ -86,6 +86,19 @@ func (lv *lookupVindex) externalize() {
 	lv.expectWriteOnly(false)
 }
 
+func (lv *lookupVindex) internalize() {
+	args := []string{
+		"LookupVindex",
+		"--name", lv.name,
+		"--table-keyspace=" + lv.ownerTableKeyspace,
+		"internalize",
+		"--keyspace=" + lv.tableKeyspace,
+	}
+	err := vc.VtctldClient.ExecuteCommand(args...)
+	require.NoError(lv.t, err, "error executing LookupVindex internalize: %v", err)
+	lv.expectWriteOnly(true)
+}
+
 func (lv *lookupVindex) show() error {
 	return nil
 }

go/test/endtoend/vreplication/lookup_vindex_test.go

@@ -84,6 +84,7 @@ type lookupTestCase struct {
 	initQuery            string
 	runningQuery         string
 	postExternalizeQuery string
+	postInternalizeQuery string
 	cleanupQuery         string
 }
 
@@ -106,6 +107,7 @@ func TestLookupVindex(t *testing.T) {
 	initQuery := "insert into t1 (c1, c2, val) values (1, 1, 'val1'), (2, 2, 'val2'), (3, 3, 'val3')"
 	runningQuery := "insert into t1 (c1, c2, val) values (4, 4, 'val4'), (5, 5, 'val5'), (6, 6, 'val6')"
 	postExternalizeQuery := "insert into t1 (c1, c2, val) values (7, 7, 'val7'), (8, 8, 'val8'), (9, 9, 'val9')"
+	postInternalizeQuery := "insert into t1 (c1, c2, val) values (10, 10, 'val10'), (11, 11, 'val11'), (12, 12, 'val12')"
 	cleanupQuery := "delete from t1"
 
 	testCases := []lookupTestCase{
@@ -158,6 +160,7 @@ func TestLookupVindex(t *testing.T) {
 			tc.initQuery = initQuery
 			tc.runningQuery = runningQuery
 			tc.postExternalizeQuery = postExternalizeQuery
+			tc.postInternalizeQuery = postInternalizeQuery
 			tc.cleanupQuery = cleanupQuery
 			testLookupVindex(t, &tc)
 		})
@@ -196,6 +199,14 @@ func testLookupVindex(t *testing.T, tc *lookupTestCase) {
 		waitForRowCount(t, vtgateConn, tc.lv.ownerTableKeyspace, lv.name, totalRows)
 	})
 
+	t.Run("internalize", func(t *testing.T) {
+		tc.lv.internalize()
+		totalRows += getNumRowsInQuery(t, tc.postInternalizeQuery)
+		_, err := vtgateConn.ExecuteFetch(tc.postInternalizeQuery, 1000, false)
+		require.NoError(t, err)
+		waitForRowCount(t, vtgateConn, tc.lv.ownerTableKeyspace, lv.name, totalRows)
+	})
+
 	t.Run("cleanup", func(t *testing.T) {
 		_, err := vtgateConn.ExecuteFetch(tc.cleanupQuery, 1000, false)
 		require.NoError(t, err)

go/test/endtoend/vreplication/vreplication_test.go

@@ -417,6 +417,14 @@ func testVreplicationWorkflows(t *testing.T, limited bool, binlogRowImage string
 		vdx = gjson.Get(customerVSchema, fmt.Sprintf("vindexes.%s", vindexName))
 		require.NotNil(t, vdx, "lookup vindex %s not found", vindexName)
 		require.NotEqual(t, "true", vdx.Get("params.write_only").String(), "did not expect write_only parameter to be true")
+
+		err = vc.VtctldClient.ExecuteCommand("LookupVindex", "--name", vindexName, "--table-keyspace=product", "internalize", "--keyspace=customer")
+		require.NoError(t, err, "error executing LookupVindex internalize: %v", err)
+		customerVSchema, err = vc.VtctldClient.ExecuteCommandWithOutput("GetVSchema", "customer")
+		require.NoError(t, err, "error executing GetVSchema: %v", err)
+		vdx = gjson.Get(customerVSchema, fmt.Sprintf("vindexes.%s", vindexName))
+		require.NotNil(t, vdx, "lookup vindex %s not found", vindexName)
+		require.Equal(t, "true", vdx.Get("params.write_only").String(), "expected write_only parameter to be true")
 	})
 }