Add Keyspace/Shard properties to VEvent

Signed-off-by: Rohit Nayak <[email protected]>
Signed-off-by: Vilius Okockis <[email protected]>

go/vt/vtgate/endtoend/vstream_test.go

@@ -340,6 +340,11 @@ func TestVStreamSharded(t *testing.T) {
 		switch err {
 		case nil:
 			for _, event := range events {
+				// check for Keyspace/Shard values first and then reset them so that we don't have to add them to each expected event proto string
+				require.Equal(t, "ks", event.Keyspace, "event has an incorrect keyspace attribute: %s", event.Keyspace)
+				require.True(t, event.Shard == "-80" || event.Shard == "80-", "event has an incorrect shard attribute: %s", event.Shard)
+				event.Keyspace = ""
+				event.Shard = ""
 				switch event.Type {
 				case binlogdatapb.VEventType_ROW, binlogdatapb.VEventType_FIELD:
 					evs = append(evs, event)

go/vt/vtgate/vstream_manager.go

@@ -604,8 +604,10 @@ func (vs *vstream) sendAll(ctx context.Context, sgtid *binlogdatapb.ShardGtid, e
 				// Update the VGtid and send that instead.
 				sgtid.Gtid = event.Gtid
 				events[j] = &binlogdatapb.VEvent{
-					Type:  binlogdatapb.VEventType_VGTID,
-					Vgtid: proto.Clone(vs.vgtid).(*binlogdatapb.VGtid),
+					Type:     binlogdatapb.VEventType_VGTID,
+					Vgtid:    proto.Clone(vs.vgtid).(*binlogdatapb.VGtid),
+					Keyspace: event.Keyspace,
+					Shard:    event.Shard,
 				}
 			} else if event.Type == binlogdatapb.VEventType_LASTPK {
 				var foundIndex = -1
@@ -631,8 +633,10 @@ func (vs *vstream) sendAll(ctx context.Context, sgtid *binlogdatapb.ShardGtid, e
 					}
 				}
 				events[j] = &binlogdatapb.VEvent{
-					Type:  binlogdatapb.VEventType_VGTID,
-					Vgtid: proto.Clone(vs.vgtid).(*binlogdatapb.VGtid),
+					Type:     binlogdatapb.VEventType_VGTID,
+					Vgtid:    proto.Clone(vs.vgtid).(*binlogdatapb.VGtid),
+					Keyspace: event.Keyspace,
+					Shard:    event.Shard,
 				}
 			}
 		}

go/vt/vtgate/vstream_manager_test.go

@@ -48,8 +48,8 @@ import (
 var mu sync.Mutex
 
 func TestVStreamSkew(t *testing.T) {
-	stream := func(conn *sandboxconn.SandboxConn, shard string, count, idx int64) {
-		vevents := getVEvents(shard, count, idx)
+	stream := func(conn *sandboxconn.SandboxConn, keyspace, shard string, count, idx int64) {
+		vevents := getVEvents(keyspace, shard, count, idx)
 		for _, ev := range vevents {
 			conn.VStreamCh <- ev
 			time.Sleep(time.Duration(idx*100) * time.Millisecond)
@@ -94,15 +94,15 @@ func TestVStreamSkew(t *testing.T) {
 				sbc0.VStreamCh = make(chan *binlogdatapb.VEvent)
 				want += 2 * tcase.numEventsPerShard
 				vgtid.ShardGtids = append(vgtid.ShardGtids, &binlogdatapb.ShardGtid{Keyspace: ks, Gtid: "pos", Shard: "-20"})
-				go stream(sbc0, "-20", tcase.numEventsPerShard, tcase.shard0idx)
+				go stream(sbc0, ks, "-20", tcase.numEventsPerShard, tcase.shard0idx)
 			}
 			if tcase.shard1idx != 0 {
 				sbc1 = hc.AddTestTablet(cell, "1.1.1.1", 1002, ks, "20-40", topodatapb.TabletType_MASTER, true, 1, nil)
 				addTabletToSandboxTopo(t, st, ks, "20-40", sbc1.Tablet())
 				sbc1.VStreamCh = make(chan *binlogdatapb.VEvent)
 				want += 2 * tcase.numEventsPerShard
 				vgtid.ShardGtids = append(vgtid.ShardGtids, &binlogdatapb.ShardGtid{Keyspace: ks, Gtid: "pos", Shard: "20-40"})
-				go stream(sbc1, "20-40", tcase.numEventsPerShard, tcase.shard1idx)
+				go stream(sbc1, ks, "20-40", tcase.numEventsPerShard, tcase.shard1idx)
 			}
 			ch := startVStream(ctx, t, vsm, vgtid, &vtgatepb.VStreamFlags{MinimizeSkew: true})
 			var receivedEvents []*binlogdatapb.VEvent
@@ -529,7 +529,10 @@ func TestVStreamJournalOneToMany(t *testing.T) {
 			}},
 		},
 	}
-	if !proto.Equal(got.Events[0], wantevent) {
+	gotEvent := got.Events[0]
+	gotEvent.Keyspace = ""
+	gotEvent.Shard = ""
+	if !proto.Equal(gotEvent, wantevent) {
 		t.Errorf("vgtid: %v, want %v", got.Events[0], wantevent)
 	}
 }
@@ -643,7 +646,10 @@ func TestVStreamJournalManyToOne(t *testing.T) {
 			}},
 		},
 	}
-	if !proto.Equal(got.Events[0], wantevent) {
+	gotEvent := got.Events[0]
+	gotEvent.Keyspace = ""
+	gotEvent.Shard = ""
+	if !proto.Equal(gotEvent, wantevent) {
 		t.Errorf("vgtid: %v, want %v", got.Events[0], wantevent)
 	}
 	verifyEvents(t, ch, want1)
@@ -1096,7 +1102,7 @@ func verifyEvents(t *testing.T, ch <-chan *binlogdatapb.VStreamResponse, wants .
 	}
 }
 
-func getVEvents(shard string, count, idx int64) []*binlogdatapb.VEvent {
+func getVEvents(keyspace, shard string, count, idx int64) []*binlogdatapb.VEvent {
 	mu.Lock()
 	defer mu.Unlock()
 	var vevents []*binlogdatapb.VEvent
@@ -1108,12 +1114,16 @@ func getVEvents(shard string, count, idx int64) []*binlogdatapb.VEvent {
 			Type: binlogdatapb.VEventType_GTID, Gtid: fmt.Sprintf("gtid-%s-%d", shard, j),
 			Timestamp:   currentTime - j,
 			CurrentTime: currentTime * 1e9,
+			Keyspace:    keyspace,
+			Shard:       shard,
 		})
 
 		vevents = append(vevents, &binlogdatapb.VEvent{
 			Type:        binlogdatapb.VEventType_COMMIT,
 			Timestamp:   currentTime - j,
 			CurrentTime: currentTime * 1e9,
+			Keyspace:    keyspace,
+			Shard:       shard,
 		})
 	}
 	return vevents

go/vt/vttablet/tabletserver/vstreamer/copy.go

@@ -131,7 +131,9 @@ func (uvs *uvstreamer) sendEventsForRows(ctx context.Context, tableName string,
 	var evs []*binlogdatapb.VEvent
 	for _, row := range rows.Rows {
 		ev := &binlogdatapb.VEvent{
-			Type: binlogdatapb.VEventType_ROW,
+			Type:     binlogdatapb.VEventType_ROW,
+			Keyspace: uvs.vse.keyspace,
+			Shard:    uvs.vse.shard,
 			RowEvent: &binlogdatapb.RowEvent{
 				TableName: tableName,
 				Keyspace:  uvs.vse.keyspace,
@@ -154,11 +156,15 @@ func (uvs *uvstreamer) sendEventsForRows(ctx context.Context, tableName string,
 
 	ev := &binlogdatapb.VEvent{
 		Type:        binlogdatapb.VEventType_LASTPK,
+		Keyspace:    uvs.vse.keyspace,
+		Shard:       uvs.vse.shard,
 		LastPKEvent: lastPKEvent,
 	}
 	evs = append(evs, ev)
 	evs = append(evs, &binlogdatapb.VEvent{
-		Type: binlogdatapb.VEventType_COMMIT,
+		Type:     binlogdatapb.VEventType_COMMIT,
+		Keyspace: uvs.vse.keyspace,
+		Shard:    uvs.vse.shard,
 	})
 
 	if err := uvs.send(evs); err != nil {

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

@@ -98,6 +98,10 @@ func newUVStreamer(ctx context.Context, vse *Engine, cp dbconfigs.Connector, se
 	}
 	send2 := func(evs []*binlogdatapb.VEvent) error {
 		vse.vstreamerEventsStreamed.Add(int64(len(evs)))
+		for _, ev := range evs {
+			ev.Keyspace = vse.keyspace
+			ev.Shard = vse.shard
+		}
 		return send(evs)
 	}
 	uvs := &uvstreamer{
@@ -464,17 +468,27 @@ func (uvs *uvstreamer) setPosition(gtid string, isInTx bool) error {
 		return nil
 	}
 	gtidEvent := &binlogdatapb.VEvent{
-		Type: binlogdatapb.VEventType_GTID,
-		Gtid: gtid,
+		Type:     binlogdatapb.VEventType_GTID,
+		Gtid:     gtid,
+		Keyspace: uvs.vse.keyspace,
+		Shard:    uvs.vse.shard,
 	}
 
 	var evs []*binlogdatapb.VEvent
 	if !isInTx {
-		evs = append(evs, &binlogdatapb.VEvent{Type: binlogdatapb.VEventType_BEGIN})
+		evs = append(evs, &binlogdatapb.VEvent{
+			Type:     binlogdatapb.VEventType_BEGIN,
+			Keyspace: uvs.vse.keyspace,
+			Shard:    uvs.vse.shard,
+		})
 	}
 	evs = append(evs, gtidEvent)
 	if !isInTx {
-		evs = append(evs, &binlogdatapb.VEvent{Type: binlogdatapb.VEventType_COMMIT})
+		evs = append(evs, &binlogdatapb.VEvent{
+			Type:     binlogdatapb.VEventType_COMMIT,
+			Keyspace: uvs.vse.keyspace,
+			Shard:    uvs.vse.shard,
+		})
 	}
 	if err := uvs.send(evs); err != nil {
 		return err

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

@@ -209,6 +209,8 @@ func (vs *vstreamer) parseEvents(ctx context.Context, events <-chan mysql.Binlog
 	// all existing rows are sent without the new row.
 	// If a single row exceeds the packet size, it will be in its own packet.
 	bufferAndTransmit := func(vevent *binlogdatapb.VEvent) error {
+		vevent.Keyspace = vs.vse.keyspace
+		vevent.Shard = vs.vse.shard
 		switch vevent.Type {
 		case binlogdatapb.VEventType_GTID, binlogdatapb.VEventType_BEGIN, binlogdatapb.VEventType_FIELD,
 			binlogdatapb.VEventType_JOURNAL:
@@ -417,7 +419,7 @@ func (vs *vstreamer) parseEvent(ev mysql.BinlogEvent) ([]*binlogdatapb.VEvent, e
 				Type: binlogdatapb.VEventType_BEGIN,
 			})
 		}
-		vs.pos = mysql.AppendGTID(vs.pos, gtid) //TODO: #sugu why Append?
+		vs.pos = mysql.AppendGTID(vs.pos, gtid)
 	case ev.IsXID():
 		vevents = append(vevents, &binlogdatapb.VEvent{
 			Type: binlogdatapb.VEventType_GTID,

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

@@ -2072,6 +2072,8 @@ func expectLog(ctx context.Context, t *testing.T, input interface{}, ch <-chan [
 		for i, want := range wantset {
 			// CurrentTime is not testable.
 			evs[i].CurrentTime = 0
+			evs[i].Keyspace = ""
+			evs[i].Shard = ""
 			switch want {
 			case "begin":
 				if evs[i].Type != binlogdatapb.VEventType_BEGIN {

proto/binlogdata.proto

@@ -375,6 +375,11 @@ message VEvent {
   int64 current_time = 20;
   // LastPK is the last PK for a table
   LastPKEvent last_p_k_event = 21;
+  // the source keyspace
+  string keyspace = 22;
+  // the source shard
+  string shard = 23;
+
 }
 
 message MinimalTable {