Run VStream copy only when VGTID requires it, use TablesToCopy in those cases (vitessio#18938)

Signed-off-by: twthorn <thomaswilliamthornton@gmail.com>
Signed-off-by: Thomas Thornton <thomaswilliamthornton@gmail.com>

Signed-off-by: Thomas Thornton <thomaswilliamthornton@gmail.com>

Author: twthorn

go/vt/vttablet/tabletserver/vstreamer/uvstreamer.go

@@ -417,10 +417,10 @@ func (uvs *uvstreamer) currentPosition() (replication.Position, error) {
 // 3. TablePKs not nil, startPos empty => table copy (for pks > lastPK)
 // 4. TablePKs not nil, startPos set => run catchup from startPos, then table copy  (for pks > lastPK)
 //
-// If TablesToCopy option is not nil, copy only the tables listed in TablesToCopy.
-// For other tables not in TablesToCopy, if startPos is set, perform catchup starting from startPos.
+// If table copy phase should run based on one of the previous states, then only copy the tables in
+// TablesToCopy list.
 func (uvs *uvstreamer) init() error {
-	if uvs.startPos == "" /* full copy */ || len(uvs.inTablePKs) > 0 /* resume copy */ || len(uvs.options.GetTablesToCopy()) > 0 /* copy specific tables */ {
+	if uvs.startPos == "" /* full copy */ || len(uvs.inTablePKs) > 0 /* resume copy */ {
 		if err := uvs.buildTablePlan(); err != nil {
 			return err
 		}

go/vt/vttablet/tabletserver/vstreamer/vstreamer_test.go

@@ -2307,3 +2307,30 @@ func TestFilteredIsNullOperator(t *testing.T) {
 		})
 	}
 }
+
+func TestUVStreamerNoCopyWithGTID(t *testing.T) {
+	execStatements(t, []string{
+		"create table t1(id int, val varchar(128), primary key(id))",
+		"insert into t1 values (1, 'val1')",
+	})
+	defer execStatements(t, []string{
+		"drop table t1",
+	})
+	ctx := context.Background()
+	filter := &binlogdatapb.Filter{
+		Rules: []*binlogdatapb.Rule{{
+			Match:  "t1",
+			Filter: "select * from t1",
+		}},
+	}
+	pos := primaryPosition(t)
+	options := &binlogdatapb.VStreamOptions{
+		TablesToCopy: []string{"t1"},
+	}
+	uvs := newUVStreamer(ctx, engine, env.Dbcfgs.DbaWithDB(), env.SchemaEngine, pos,
+		nil, filter, testLocalVSchema, throttlerapp.VStreamerName,
+		func([]*binlogdatapb.VEvent) error { return nil }, options)
+	err := uvs.init()
+	require.NoError(t, err)
+	require.Empty(t, uvs.plans, "Should not build table plans when startPos is set")
+}