fix: Avoid creation of workflows with non-empty tables in target keyspace

go/vt/vtctl/workflow/materializer.go

@@ -290,6 +290,15 @@ func (mz *materializer) deploySchema() error {
 		}
 	}
 
+	// Check if any table being moved is already non-empty in the target keyspace.
+	// Skip this check for multi-tenant migrations.
+	if !mz.IsMultiTenantMigration() {
+		err := validateEmptyTables(mz.ctx, mz.ts, mz.tmc, mz.targetShards, mz.ms.TableSettings)
+		if err != nil {
+			return err
+		}
+	}
+
 	err := forAllShards(mz.targetShards, func(target *topo.ShardInfo) error {
 		allTables := []string{"/.*/"}
 

go/vt/vtctl/workflow/materializer_test.go

@@ -51,6 +51,7 @@ const (
 	mzGetCopyState       = "select distinct table_name from _vt.copy_state cs, _vt.vreplication vr where vr.id = cs.vrepl_id and vr.id = 1"
 	mzGetLatestCopyState = "select vrepl_id, table_name, lastpk from _vt.copy_state where vrepl_id in (1) and id in (select max(id) from _vt.copy_state where vrepl_id in (1) group by vrepl_id, table_name)"
 	insertPrefix         = `/insert into _vt.vreplication\(workflow, source, pos, max_tps, max_replication_lag, cell, tablet_types, time_updated, transaction_timestamp, state, db_name, workflow_type, workflow_sub_type, defer_secondary_keys, options\) values `
+	getNonEmptyTable     = "select 1 from `t1` limit 1"
 )
 
 var (
@@ -520,6 +521,7 @@ func TestMigrateVSchema(t *testing.T) {
 	defer env.close()
 
 	env.tmc.expectVRQuery(100, mzCheckJournal, &sqltypes.Result{})
+	env.tmc.expectVRQuery(200, getNonEmptyTable, &sqltypes.Result{})
 	env.tmc.expectVRQuery(200, mzGetCopyState, &sqltypes.Result{})
 	env.tmc.expectVRQuery(200, mzGetLatestCopyState, &sqltypes.Result{})
 
@@ -578,6 +580,7 @@ func TestMoveTablesDDLFlag(t *testing.T) {
 			// a circular dependency.
 			// The TabletManager portion is tested in rpc_vreplication_test.go.
 			env.tmc.expectVRQuery(100, mzCheckJournal, &sqltypes.Result{})
+			env.tmc.expectVRQuery(200, getNonEmptyTable, &sqltypes.Result{})
 			env.tmc.expectVRQuery(200, mzGetCopyState, &sqltypes.Result{})
 			env.tmc.expectVRQuery(200, mzGetLatestCopyState, &sqltypes.Result{})
 
@@ -610,6 +613,7 @@ func TestMoveTablesDDLFlag(t *testing.T) {
 func TestShardedAutoIncHandling(t *testing.T) {
 	tableName := "t1"
 	tableDDL := fmt.Sprintf("create table %s (id int not null auto_increment primary key, c1 varchar(10))", tableName)
+	validateEmptyTableQuery := fmt.Sprintf("select 1 from `%s` limit 1", tableName)
 	ms := &vtctldatapb.MaterializeSettings{
 		Workflow:       "workflow",
 		SourceKeyspace: "sourceks",
@@ -705,6 +709,7 @@ func TestShardedAutoIncHandling(t *testing.T) {
 				},
 			},
 			expectQueries: []string{
+				validateEmptyTableQuery,
 				tableDDL, // Unchanged
 			},
 		},
@@ -750,6 +755,7 @@ func TestShardedAutoIncHandling(t *testing.T) {
 				},
 			},
 			expectQueries: []string{ // auto_increment clause removed
+				validateEmptyTableQuery,
 				fmt.Sprintf(`create table %s (
 	id int not null primary key,
 	c1 varchar(10)
@@ -806,6 +812,7 @@ func TestShardedAutoIncHandling(t *testing.T) {
 				},
 			},
 			expectQueries: []string{ // auto_increment clause removed
+				validateEmptyTableQuery,
 				fmt.Sprintf(`create table %s (
 	id int not null primary key,
 	c1 varchar(10)
@@ -858,6 +865,7 @@ func TestShardedAutoIncHandling(t *testing.T) {
 				},
 			},
 			expectQueries: []string{ // auto_increment clause removed
+				validateEmptyTableQuery,
 				fmt.Sprintf(`create table %s (
 	id int not null primary key,
 	c1 varchar(10)
@@ -943,6 +951,7 @@ func TestMoveTablesNoRoutingRules(t *testing.T) {
 	// a circular dependency.
 	// The TabletManager portion is tested in rpc_vreplication_test.go.
 	env.tmc.expectVRQuery(100, mzCheckJournal, &sqltypes.Result{})
+	env.tmc.expectVRQuery(200, getNonEmptyTable, &sqltypes.Result{})
 	env.tmc.expectVRQuery(200, mzGetCopyState, &sqltypes.Result{})
 	env.tmc.expectVRQuery(200, mzGetLatestCopyState, &sqltypes.Result{})
 
@@ -1061,6 +1070,7 @@ func TestCreateLookupVindexFull(t *testing.T) {
 	}
 
 	env.tmc.expectVRQuery(100, mzCheckJournal, &sqltypes.Result{})
+	env.tmc.expectVRQuery(200, "select 1 from `lookup` limit 1", &sqltypes.Result{})
 	env.tmc.expectVRQuery(200, "/CREATE TABLE `lookup`", &sqltypes.Result{})
 	env.tmc.expectVRQuery(200, mzGetCopyState, &sqltypes.Result{})
 	env.tmc.expectVRQuery(200, mzGetLatestCopyState, &sqltypes.Result{})
@@ -2777,7 +2787,10 @@ func TestCreateLookupVindexFailures(t *testing.T) {
 					},
 				},
 			},
-			vrepExecQueries: []string{"CREATE TABLE `t1_lkp` (\n`c1` INT,\n  `keyspace_id` varbinary(128),\n  PRIMARY KEY (`c1`)\n)"},
+			vrepExecQueries: []string{
+				"select 1 from `t1_lkp` limit 1",
+				"CREATE TABLE `t1_lkp` (\n`c1` INT,\n  `keyspace_id` varbinary(128),\n  PRIMARY KEY (`c1`)\n)",
+			},
 			createRequest: &createVReplicationWorkflowRequestResponse{
 				req: nil, // We don't care about defining it in this case
 				res: &tabletmanagerdatapb.CreateVReplicationWorkflowResponse{},
@@ -3096,6 +3109,7 @@ func TestKeyRangesEqualOptimization(t *testing.T) {
 				if tablet.Keyspace != targetKs || tablet.Type != topodatapb.TabletType_PRIMARY {
 					continue
 				}
+				env.tmc.expectVRQuery(int(tablet.Alias.Uid), getNonEmptyTable, &sqltypes.Result{})
 				// If we are doing a partial MoveTables, we will only perform the workflow
 				// stream creation / INSERT statment on the shard(s) we're migrating.
 				if len(tc.moveTablesReq.SourceShards) > 0 && !slices.Contains(tc.moveTablesReq.SourceShards, tablet.Shard) {

go/vt/vtctl/workflow/utils.go

@@ -28,6 +28,8 @@ import (
 	"strings"
 	"sync"
 
+	"golang.org/x/exp/maps"
+	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/encoding/prototext"
 
 	"vitess.io/vitess/go/mysql/sqlerror"
@@ -1019,3 +1021,59 @@ func applyTargetShards(ts *trafficSwitcher, targetShards []string) error {
 	}
 	return nil
 }
+
+// validateEmptyTables checks if all specified tables are empty across specified shards.
+// It queries each shard's primary tablet and if any non-empty table is found, it returns an error
+// containing a list of non-empty tables.
+func validateEmptyTables(ctx context.Context, ts *topo.Server, tmc tmclient.TabletManagerClient, shards []*topo.ShardInfo, tableSettings []*vtctldatapb.TableMaterializeSettings) error {
+	var mu sync.Mutex
+	isTableFaulty := map[string]bool{}
+
+	err := forAllShards(shards, func(shard *topo.ShardInfo) error {
+		primary := shard.PrimaryAlias
+		if primary == nil {
+			return fmt.Errorf("shard does not have a primary: %v", shard.ShardName())
+		}
+
+		ti, err := ts.GetTablet(ctx, primary)
+		if err != nil {
+			return err
+		}
+
+		eg, groupCtx := errgroup.WithContext(ctx)
+		eg.SetLimit(20)
+
+		for _, ts := range tableSettings {
+			eg.Go(func() error {
+				query := fmt.Sprintf("select 1 from `%s` limit 1", ts.TargetTable)
+				res, err := tmc.ExecuteFetchAsDba(groupCtx, ti.Tablet, true, &tabletmanagerdatapb.ExecuteFetchAsDbaRequest{
+					Query:   []byte(query),
+					MaxRows: 1,
+				})
+				// Ignore table not found error
+				if err != nil && !IsTableDidNotExistError(err) {
+					return err
+				}
+				if res != nil && len(res.Rows) > 0 {
+					mu.Lock()
+					isTableFaulty[ts.TargetTable] = true
+					mu.Unlock()
+				}
+				return nil
+			})
+		}
+		if err = eg.Wait(); err != nil {
+			return err
+		}
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	faultyTables := maps.Keys(isTableFaulty)
+	if len(faultyTables) > 0 {
+		return fmt.Errorf("non-empty tables found in target keyspace: %s", strings.Join(faultyTables, ", "))
+	}
+	return nil
+}

go/vt/vtctl/workflow/utils_test.go

@@ -11,16 +11,21 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	clientv3 "go.etcd.io/etcd/client/v3"
 
 	"vitess.io/vitess/go/testfiles"
 	"vitess.io/vitess/go/vt/log"
-	"vitess.io/vitess/go/vt/proto/vtctldata"
 	"vitess.io/vitess/go/vt/topo"
 	"vitess.io/vitess/go/vt/topo/etcd2topo"
 	"vitess.io/vitess/go/vt/topo/memorytopo"
 	"vitess.io/vitess/go/vt/topotools"
+	"vitess.io/vitess/go/vt/vtctl/grpcvtctldserver/testutil"
+
+	querypb "vitess.io/vitess/go/vt/proto/query"
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
+	"vitess.io/vitess/go/vt/proto/vtctldata"
 )
 
 // TestCreateDefaultShardRoutingRules confirms that the default shard routing rules are created correctly for sharded
@@ -243,3 +248,121 @@ func startEtcd(t *testing.T) string {
 
 	return clientAddr
 }
+
+func TestValidateEmptyTables(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	ts := memorytopo.NewServer(ctx, "zone1")
+	defer ts.Close()
+
+	ks := "test_keyspace"
+	shard1 := "-40"
+	shard2 := "40-80"
+	shard3 := "80-"
+	err := ts.CreateKeyspace(ctx, ks, &topodatapb.Keyspace{})
+	require.NoError(t, err)
+
+	err = ts.CreateShard(ctx, ks, shard1)
+	require.NoError(t, err)
+	err = ts.CreateShard(ctx, ks, shard2)
+	require.NoError(t, err)
+	err = ts.CreateShard(ctx, ks, shard3)
+	require.NoError(t, err)
+
+	tablet1 := &topodatapb.Tablet{
+		Alias: &topodatapb.TabletAlias{
+			Cell: "zone1",
+			Uid:  100,
+		},
+		Keyspace: ks,
+		Shard:    shard1,
+	}
+	tablet2 := &topodatapb.Tablet{
+		Alias: &topodatapb.TabletAlias{
+			Cell: "zone1",
+			Uid:  200,
+		},
+		Keyspace: ks,
+		Shard:    shard2,
+	}
+	tablet3 := &topodatapb.Tablet{
+		Alias: &topodatapb.TabletAlias{
+			Cell: "zone1",
+			Uid:  300,
+		},
+		Keyspace: ks,
+		Shard:    shard3,
+	}
+	err = ts.CreateTablet(ctx, tablet1)
+	require.NoError(t, err)
+	err = ts.CreateTablet(ctx, tablet2)
+	require.NoError(t, err)
+	err = ts.CreateTablet(ctx, tablet3)
+	require.NoError(t, err)
+
+	s1, err := ts.UpdateShardFields(ctx, ks, shard1, func(si *topo.ShardInfo) error {
+		si.Shard.PrimaryAlias = tablet1.Alias
+		return nil
+	})
+	require.NoError(t, err)
+	s2, err := ts.UpdateShardFields(ctx, ks, shard2, func(si *topo.ShardInfo) error {
+		si.Shard.PrimaryAlias = tablet2.Alias
+		return nil
+	})
+	require.NoError(t, err)
+	s3, err := ts.UpdateShardFields(ctx, ks, shard3, func(si *topo.ShardInfo) error {
+		si.Shard.PrimaryAlias = tablet3.Alias
+		return nil
+	})
+	require.NoError(t, err)
+
+	tmc := &testutil.TabletManagerClient{
+		ExecuteFetchAsDbaResults: map[string]struct {
+			Response *querypb.QueryResult
+			Error    error
+		}{
+			"zone1-0000000100": {
+				Response: &querypb.QueryResult{
+					Rows: []*querypb.Row{{
+						Lengths: []int64{1},
+						Values:  []byte("1"),
+					},
+					},
+				},
+			},
+			"zone1-0000000200": {
+				Response: &querypb.QueryResult{
+					Rows: []*querypb.Row{{
+						Lengths: []int64{1},
+						Values:  []byte("1"),
+					},
+					},
+				},
+			},
+			"zone1-0000000300": {
+				Response: &querypb.QueryResult{
+					Rows: []*querypb.Row{{}},
+				},
+			},
+		},
+	}
+
+	tableSettings := []*vtctldata.TableMaterializeSettings{
+		{
+			TargetTable: "table1",
+		},
+		{
+			TargetTable: "table2",
+		},
+		{
+			TargetTable: "table3",
+		},
+	}
+	err = validateEmptyTables(ctx, ts, tmc, []*topo.ShardInfo{s1, s2, s3}, tableSettings)
+	assert.ErrorContains(t, err, "table1")
+	assert.ErrorContains(t, err, "table2")
+	assert.ErrorContains(t, err, "table3")
+
+	err = validateEmptyTables(ctx, ts, tmc, []*topo.ShardInfo{s1, s2, s3}, []*vtctldata.TableMaterializeSettings{})
+	assert.NoError(t, err, "should not throw any error for empty table settings slice")
+}

go/vt/vttablet/tabletmanager/framework_test.go

@@ -480,8 +480,8 @@ func (tmc *fakeTMClient) VReplicationExec(ctx context.Context, tablet *topodatap
 	}
 	for qry, res := range tmc.vreQueries[int(tablet.Alias.Uid)] {
 		if strings.HasPrefix(qry, "/") {
-			re := regexp.MustCompile(qry)
-			if re.MatchString(qry) {
+			re := regexp.MustCompile(qry[1:])
+			if re.MatchString(query) {
 				return res, nil
 			}
 		}