Add a flag to vstream to exclude keyspace from table name (#18274)

go/vt/vtgate/vstream_manager.go

@@ -685,16 +685,10 @@ func (vs *vstream) streamFromTablet(ctx context.Context, sgtid *binlogdatapb.Sha
 			for i, event := range events {
 				switch event.Type {
 				case binlogdatapb.VEventType_FIELD:
-					// Update table names and send.
-					// If we're streaming from multiple keyspaces, this will disambiguate
-					// duplicate table names.
-					ev := event.CloneVT()
-					ev.FieldEvent.TableName = sgtid.Keyspace + "." + ev.FieldEvent.TableName
+					ev := maybeUpdateTableName(event, sgtid.Keyspace, vs.flags.GetExcludeKeyspaceFromTableName(), extractFieldTableName)
 					sendevents = append(sendevents, ev)
 				case binlogdatapb.VEventType_ROW:
-					// Update table names and send.
-					ev := event.CloneVT()
-					ev.RowEvent.TableName = sgtid.Keyspace + "." + ev.RowEvent.TableName
+					ev := maybeUpdateTableName(event, sgtid.Keyspace, vs.flags.GetExcludeKeyspaceFromTableName(), extractRowTableName)
 					sendevents = append(sendevents, ev)
 				case binlogdatapb.VEventType_COMMIT, binlogdatapb.VEventType_DDL, binlogdatapb.VEventType_OTHER:
 					sendevents = append(sendevents, event)
@@ -833,6 +827,29 @@ func (vs *vstream) streamFromTablet(ctx context.Context, sgtid *binlogdatapb.Sha
 
 }
 
+// maybeUpdateTableNames updates table names when the ExcludeKeyspaceFromTableName flag is disabled.
+// If we're streaming from multiple keyspaces, updating the table names by inserting the keyspace will disambiguate
+// duplicate table names. If we enable the ExcludeKeyspaceFromTableName flag to not update the table names, there is no need to
+// clone the entire event, whcih improves performance. This is typically safely used by clients only streaming one keyspace.
+func maybeUpdateTableName(event *binlogdatapb.VEvent, keyspace string, excludeKeyspaceFromTableName bool,
+	tableNameExtractor func(ev *binlogdatapb.VEvent) *string) *binlogdatapb.VEvent {
+	if excludeKeyspaceFromTableName {
+		return event
+	}
+	ev := event.CloneVT()
+	tableName := tableNameExtractor(ev)
+	*tableName = keyspace + "." + *tableName
+	return ev
+}
+
+func extractFieldTableName(ev *binlogdatapb.VEvent) *string {
+	return &ev.FieldEvent.TableName
+}
+
+func extractRowTableName(ev *binlogdatapb.VEvent) *string {
+	return &ev.RowEvent.TableName
+}
+
 // shouldRetry determines whether we should exit immediately or retry the vstream.
 // The first return value determines if the error can be retried, while the second
 // indicates whether the tablet with which the error occurred should be omitted