Remove Usage of VReplicationExec For _vt.vreplication Reads

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

@@ -93,6 +93,7 @@
 
 	ms := &vtctldatapb.MaterializeSettings{
 		Workflow:                  common.BaseOptions.Workflow,
+		MaterializationIntent:     vtctldatapb.MaterializationIntent_CUSTOM,
 		TargetKeyspace:            common.BaseOptions.TargetKeyspace,
 		SourceKeyspace:            createOptions.SourceKeyspace,
 		TableSettings:             createOptions.TableSettings.val,

go/test/endtoend/vreplication/helper_test.go

@@ -568,11 +568,27 @@ func isTableInDenyList(t *testing.T, vc *VitessCluster, ksShard string, table st
 	return found, nil
 }
 
-func expectNumberOfStreams(t *testing.T, vtgateConn *mysql.Conn, name string, workflow string, database string, want int) {
-	query := sqlparser.BuildParsedQuery("select count(*) from %s.vreplication where workflow='%s'", sidecarDBIdentifier, workflow).Query
+// expectNumberOfStreams waits for the given number of streams to be present and
+// by default RUNNING. If you want to wait for different states, then you can
+// pass in the state(s) you want to wait for.
+func expectNumberOfStreams(t *testing.T, vtgateConn *mysql.Conn, name string, workflow string, database string, want int, states ...string) {
+	var query string
+	if len(states) == 0 {
+		states = append(states, binlogdatapb.VReplicationWorkflowState_Running.String())
+	}
+	query = sqlparser.BuildParsedQuery("select count(*) from %s.vreplication where workflow='%s' and state in ('%s')",
+		sidecarDBIdentifier, workflow, strings.Join(states, "','")).Query
 	waitForQueryResult(t, vtgateConn, database, query, fmt.Sprintf(`[[INT64(%d)]]`, want))
 }
 
+// confirmAllStreamsRunning confirms that all of the migrated streams are running
+// after a Reshard.
+func confirmAllStreamsRunning(t *testing.T, vtgateConn *mysql.Conn, database string) {
+	query := sqlparser.BuildParsedQuery("select count(*) from %s.vreplication where state != '%s'",
+		sidecarDBIdentifier, binlogdatapb.VReplicationWorkflowState_Running.String()).Query
+	waitForQueryResult(t, vtgateConn, database, query, `[[INT64(0)]]`)
+}
+
 func printShardPositions(vc *VitessCluster, ksShards []string) {
 	for _, ksShard := range ksShards {
 		output, err := vc.VtctlClient.ExecuteCommandWithOutput("ShardReplicationPositions", ksShard)

go/test/endtoend/vreplication/migrate_test.go

@@ -128,7 +128,7 @@ func TestVtctlMigrate(t *testing.T) {
 			"--source=ext1.rating", "create", ksWorkflow); err != nil {
 			t.Fatalf("Migrate command failed with %+v : %s\n", err, output)
 		}
-		expectNumberOfStreams(t, vtgateConn, "migrate", "e1", "product:0", 1)
+		expectNumberOfStreams(t, vtgateConn, "migrate", "e1", "product:0", 1, binlogdatapb.VReplicationWorkflowState_Stopped.String())
 		waitForRowCount(t, vtgateConn, "product:0", "rating", 0)
 		waitForRowCount(t, vtgateConn, "product:0", "review", 0)
 		if output, err = vc.VtctlClient.ExecuteCommandWithOutput("Migrate", "cancel", ksWorkflow); err != nil {
@@ -267,7 +267,7 @@ func TestVtctldMigrate(t *testing.T) {
 			"--mount-name", "ext1", "--all-tables", "--auto-start=false", "--cells=extcell1")
 		require.NoError(t, err, "Migrate command failed with %s", output)
 
-		expectNumberOfStreams(t, vtgateConn, "migrate", "e1", "product:0", 1)
+		expectNumberOfStreams(t, vtgateConn, "migrate", "e1", "product:0", 1, binlogdatapb.VReplicationWorkflowState_Stopped.String())
 		waitForRowCount(t, vtgateConn, "product:0", "rating", 0)
 		waitForRowCount(t, vtgateConn, "product:0", "review", 0)
 		output, err = vc.VtctldClient.ExecuteCommandWithOutput("Migrate",

go/test/endtoend/vreplication/vreplication_test.go

@@ -338,10 +338,13 @@ func testVreplicationWorkflows(t *testing.T, limited bool, binlogRowImage string
 
 	insertMoreCustomers(t, 16)
 	reshardCustomer2to4Split(t, nil, "")
+	confirmAllStreamsRunning(t, vtgateConn, "customer:-40")
 	expectNumberOfStreams(t, vtgateConn, "Customer2to4", "sales", "product:0", 4)
 	reshardCustomer3to2SplitMerge(t)
+	confirmAllStreamsRunning(t, vtgateConn, "customer:-60")
 	expectNumberOfStreams(t, vtgateConn, "Customer3to2", "sales", "product:0", 3)
 	reshardCustomer3to1Merge(t)
+	confirmAllStreamsRunning(t, vtgateConn, "customer:0")
 	expectNumberOfStreams(t, vtgateConn, "Customer3to1", "sales", "product:0", 1)
 
 	t.Run("Verify CopyState Is Optimized Afterwards", func(t *testing.T) {

go/vt/binlog/binlogplayer/binlog_player.go

@@ -558,8 +558,7 @@ type VRSettings struct {
 	DeferSecondaryKeys bool
 }
 
-// ReadVRSettings retrieves the throttler settings for
-// vreplication from the checkpoint table.
+// ReadVRSettings retrieves the settings for a vreplication stream.
 func ReadVRSettings(dbClient DBClient, uid int32) (VRSettings, error) {
 	query := fmt.Sprintf("select pos, stop_pos, max_tps, max_replication_lag, state, workflow_type, workflow, workflow_sub_type, defer_secondary_keys from _vt.vreplication where id=%v", uid)
 	qr, err := dbClient.ExecuteFetch(query, 1)

go/vt/binlog/binlogplayer/mock_dbclient.go

@@ -182,6 +182,10 @@ func (dc *MockDBClient) Close() {
 
 // ExecuteFetch is part of the DBClient interface
 func (dc *MockDBClient) ExecuteFetch(query string, maxrows int) (qr *sqltypes.Result, err error) {
+	// Serialize ExecuteFetch to enforce a strict order on shared dbClients.
+	dc.expectMu.Lock()
+	defer dc.expectMu.Unlock()
+
 	dc.t.Helper()
 	msg := "DBClient query: %v"
 	if dc.Tag != "" {
@@ -195,8 +199,6 @@ func (dc *MockDBClient) ExecuteFetch(query string, maxrows int) (qr *sqltypes.Re
 		}
 	}
 
-	dc.expectMu.Lock()
-	defer dc.expectMu.Unlock()
 	if dc.currentResult >= len(dc.expect) {
 		msg := "DBClientMock: query: %s, no more requests are expected"
 		if dc.Tag != "" {

go/vt/discovery/tablet_picker.go

@@ -35,6 +35,7 @@
 	"vitess.io/vitess/go/vt/vttablet/tabletconn"
 
 	querypb "vitess.io/vitess/go/vt/proto/query"
+	tabletmanagerdatapb "vitess.io/vitess/go/vt/proto/tabletmanagerdata"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 )
@@ -75,6 +76,16 @@
 	}
 )
 
+// BuildTabletTypesString is a helper to build a serialized string representation of
+// the tablet type(s) and optional in order clause for later use with the TabletPicker.
+func BuildTabletTypesString(tabletTypes []topodatapb.TabletType, tabletSelectionPreference tabletmanagerdatapb.TabletSelectionPreference) string {
+	tabletTypesStr := topoproto.MakeStringTypeCSV(tabletTypes)
+	if tabletSelectionPreference == tabletmanagerdatapb.TabletSelectionPreference_INORDER {
+		tabletTypesStr = InOrderHint + tabletTypesStr
+	}
+	return tabletTypesStr
+}
+
 // GetTabletPickerRetryDelay synchronizes changes to tabletPickerRetryDelay. Used in tests only at the moment
 func GetTabletPickerRetryDelay() time.Duration {
 	muTabletPickerRetryDelay.Lock()