[release-24.0] VReplication: LookupVindex streams must not follow TABLES journals (#20917) (#20937)

Signed-off-by: Matt Lord <mattalord@gmail.com>
Co-authored-by: vitess-bot[bot] <108069721+vitess-bot[bot]@users.noreply.github.com>
Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>

Author: vitess-bot[bot]

go/vt/vttablet/tabletmanager/vreplication/engine.go

@@ -582,6 +582,15 @@ func (vre *Engine) registerJournal(journal *binlogdatapb.Journal, id int32) erro
 	}
 
 	workflow := vre.controllers[id].workflow
+	// Insurance against callers that bypass the vplayer's gate (see the
+	// VEventType_JOURNAL handling): lookup vindex backfill streams must
+	// never be transitioned by a TABLES journal, as their keyspace_id()
+	// filter is not expressible in another keyspace. transitionJournal
+	// would otherwise irrecoverably delete the streams.
+	if journal.MigrationType == binlogdatapb.MigrationType_TABLES &&
+		binlogdatapb.VReplicationWorkflowType(vre.controllers[id].workflowType) == binlogdatapb.VReplicationWorkflowType_CreateLookupIndex {
+		return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "cannot follow TABLES journal %d for lookup vindex workflow %s: lookup backfill streams cannot be retargeted to another keyspace", journal.Id, workflow)
+	}
 	key := fmt.Sprintf("%s:%d", workflow, journal.Id)
 	ks := fmt.Sprintf("%s:%s", vre.controllers[id].source.Keyspace, vre.controllers[id].source.Shard)
 	log.Info(fmt.Sprintf("Journal encountered for (%s %s): %v", key, ks, journal))

go/vt/vttablet/tabletmanager/vreplication/journal_test.go

@@ -20,6 +20,9 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
 	qh "vitess.io/vitess/go/vt/vttablet/tabletmanager/vreplication/queryhistory"
 )
@@ -314,3 +317,140 @@ func TestJournalTableMixed(t *testing.T) {
 	deleteAllVReplicationStreams(t)
 	expectDeleteQueries(t)
 }
+
+// TestJournalTablesLookupVindexIgnored guards against resharding journals
+// written by MoveTables SwitchTraffic/ReverseTraffic destroying lookup
+// vindex backfill streams (https://github.com/vitessio/vitess/issues/20915).
+// A CreateLookupIndex stream must ignore a TABLES journal -- its
+// keyspace_id() filter cannot be planned in another keyspace -- and keep
+// replicating from its current source, which the paired workflow keeps
+// feeding after the switch. SHARDS (Reshard) journals are still followed;
+// that behavior is pinned by TestJournalOneToOne/TestJournalOneToMany.
+func TestJournalTablesLookupVindexIgnored(t *testing.T) {
+	if runNoBlobTest {
+		t.Skip("CreateLookupIndex workflows do not support binlog_row_image=noblob")
+	}
+	// This test's teardown uses defer, not t.Cleanup: execStatements and
+	// the other framework helpers run on t.Context(), which is already
+	// canceled by the time t.Cleanup callbacks execute.
+	defer deleteTablet(addTablet(100))
+	defer deleteTablet(addOtherTablet(101, "other_keyspace", "0"))
+
+	execStatements(t, []string{
+		"create table t(id int, val varbinary(128), primary key(id))",
+		fmt.Sprintf("create table %s.t(id int, val varbinary(128), primary key(id))", vrepldb),
+	})
+	defer execStatements(t, []string{
+		"drop table t",
+		fmt.Sprintf("drop table %s.t", vrepldb),
+	})
+
+	filter := &binlogdatapb.Filter{
+		Rules: []*binlogdatapb.Rule{{
+			Match: "t",
+		}},
+	}
+	bls := &binlogdatapb.BinlogSource{
+		Keyspace: env.KeyspaceName,
+		Shard:    env.ShardName,
+		Filter:   filter,
+		OnDdl:    binlogdatapb.OnDDLAction_IGNORE,
+	}
+	cancel, _ := startVReplicationWithWorkflowType(t, bls, "", binlogdatapb.VReplicationWorkflowType_CreateLookupIndex)
+	defer cancel()
+
+	journal := &binlogdatapb.Journal{
+		Id:            1,
+		MigrationType: binlogdatapb.MigrationType_TABLES,
+		Participants: []*binlogdatapb.KeyspaceShard{{
+			Keyspace: "vttest",
+			Shard:    "0",
+		}},
+		Tables: []string{"t"},
+		ShardGtids: []*binlogdatapb.ShardGtid{{
+			Keyspace: "other_keyspace",
+			Shard:    "0",
+			Gtid:     "MySQL56/7b04699f-f5e9-11e9-bf88-9cb6d089e1c3:1-10",
+		}},
+	}
+	query := fmt.Sprintf("insert into _vt.resharding_journal(id, db_name, val) values (1, 'vttest', %v)", encodeString(journal.String()))
+	execStatements(t, []string{query})
+	defer execStatements(t, []string{"delete from _vt.resharding_journal"})
+
+	// The journal must be ignored: no stream deletion/recreation. The
+	// stream must still be replicating from the original source, which
+	// this insert proves end to end.
+	execStatements(t, []string{"insert into t values(1, 'aaa')"})
+	expectDBClientQueries(t, qh.Expect(
+		"begin",
+		"insert into t(id,val) values (1,_binary'aaa')",
+		"/update _vt.vreplication set pos=",
+		"commit",
+	))
+}
+
+// TestJournalRegisterLookupVindexRefused guards the engine-side insurance
+// for https://github.com/vitessio/vitess/issues/20915: even if a future
+// caller bypasses the vplayer's TABLES-journal gate, registerJournal must
+// refuse to transition a lookup vindex workflow rather than let
+// transitionJournal destroy its streams.
+func TestJournalRegisterLookupVindexRefused(t *testing.T) {
+	if runNoBlobTest {
+		t.Skip("CreateLookupIndex workflows do not support binlog_row_image=noblob")
+	}
+	// This test's teardown uses defer, not t.Cleanup: execStatements and
+	// the other framework helpers run on t.Context(), which is already
+	// canceled by the time t.Cleanup callbacks execute.
+	defer deleteTablet(addTablet(100))
+	defer deleteTablet(addOtherTablet(101, "other_keyspace", "0"))
+
+	execStatements(t, []string{
+		"create table t(id int, val varbinary(128), primary key(id))",
+		fmt.Sprintf("create table %s.t(id int, val varbinary(128), primary key(id))", vrepldb),
+	})
+	defer execStatements(t, []string{
+		"drop table t",
+		fmt.Sprintf("drop table %s.t", vrepldb),
+	})
+
+	filter := &binlogdatapb.Filter{
+		Rules: []*binlogdatapb.Rule{{
+			Match: "t",
+		}},
+	}
+	bls := &binlogdatapb.BinlogSource{
+		Keyspace: env.KeyspaceName,
+		Shard:    env.ShardName,
+		Filter:   filter,
+		OnDdl:    binlogdatapb.OnDDLAction_IGNORE,
+	}
+	cancel, id := startVReplicationWithWorkflowType(t, bls, "", binlogdatapb.VReplicationWorkflowType_CreateLookupIndex)
+	defer cancel()
+
+	journal := &binlogdatapb.Journal{
+		Id:            2,
+		MigrationType: binlogdatapb.MigrationType_TABLES,
+		Participants: []*binlogdatapb.KeyspaceShard{{
+			Keyspace: "vttest",
+			Shard:    "0",
+		}},
+		Tables: []string{"t"},
+		ShardGtids: []*binlogdatapb.ShardGtid{{
+			Keyspace: "other_keyspace",
+			Shard:    "0",
+			Gtid:     "MySQL56/7b04699f-f5e9-11e9-bf88-9cb6d089e1c3:1-10",
+		}},
+	}
+
+	err := playerEngine.registerJournal(journal, int32(id))
+	require.ErrorContains(t, err, "lookup vindex")
+
+	// The stream must be untouched: no journaler entry queued, no
+	// transition, the controller still registered.
+	playerEngine.mu.Lock()
+	_, ok := playerEngine.controllers[int32(id)]
+	journalerEmpty := len(playerEngine.journaler) == 0
+	playerEngine.mu.Unlock()
+	assert.True(t, ok, "controller was removed")
+	assert.True(t, journalerEmpty, "journaler entry was created")
+}

go/vt/vttablet/tabletmanager/vreplication/vplayer.go

@@ -852,6 +852,29 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 		case binlogdatapb.MigrationType_SHARDS:
 			// All tables of the source were migrated. So, no validation needed.
 		case binlogdatapb.MigrationType_TABLES:
+			// Lookup vindex backfill streams must never follow a TABLES
+			// journal (written by MoveTables SwitchTraffic/ReverseTraffic):
+			// their filter selects keyspace_id(), which the vstreamer
+			// resolves against the serving keyspace's vschema, so a
+			// relocated stream either fails to plan or computes wrong
+			// keyspace_id values and corrupts the lookup table. The
+			// stream's current source keeps receiving the owner table's
+			// writes via the paired workflow after the switch -- when
+			// reverse replication is running, which is the default -- so
+			// we ignore the journal and keep replicating from it. With
+			// --enable-reverse-replication=false nothing feeds the
+			// current source and the backfill goes stale until reverse
+			// replication is started or the switch is reversed. SHARDS
+			// (Reshard) journals are still followed: the keyspace -- and
+			// with it the filter's validity -- does not change.
+			if binlogdatapb.VReplicationWorkflowType(vp.vr.WorkflowType) == binlogdatapb.VReplicationWorkflowType_CreateLookupIndex {
+				log.Info("Ignoring TABLES journal for lookup vindex workflow",
+					slog.String("workflow", vp.vr.WorkflowName),
+					slog.Int64("journal_id", event.Journal.Id),
+					slog.Any("journal_tables", event.Journal.Tables),
+				)
+				return nil
+			}
 			// Validate that all or none of the tables are in the journal.
 			jtables := make(map[string]bool)
 			for _, table := range event.Journal.Tables {

go/vt/vttablet/tabletmanager/vreplication/vplayer_flaky_test.go

@@ -4230,12 +4230,19 @@ func expectJSON(t *testing.T, table string, values [][]string, id int, exec func
 
 func startVReplication(t *testing.T, bls *binlogdatapb.BinlogSource, pos string) (cancelFunc func(), id int) {
 	t.Helper()
+	// fake workflow type as MoveTables so that we can test with "noblob" binlog row image
+	return startVReplicationWithWorkflowType(t, bls, pos, binlogdatapb.VReplicationWorkflowType_MoveTables)
+}
+
+func startVReplicationWithWorkflowType(t *testing.T, bls *binlogdatapb.BinlogSource, pos string,
+	workflowType binlogdatapb.VReplicationWorkflowType,
+) (cancelFunc func(), id int) {
+	t.Helper()
 
 	if pos == "" {
 		pos = primaryPosition(t)
 	}
-	// fake workflow type as MoveTables so that we can test with "noblob" binlog row image
-	query := binlogplayer.CreateVReplication("test", bls, pos, 9223372036854775807, 9223372036854775807, 0, vrepldb, binlogdatapb.VReplicationWorkflowType_MoveTables, 0, false)
+	query := binlogplayer.CreateVReplication("test", bls, pos, 9223372036854775807, 9223372036854775807, 0, vrepldb, workflowType, 0, false)
 	qr, err := playerEngine.Exec(query)
 	require.NoError(t, err)
 	expectDBClientQueries(t, qh.Expect(