Merge pull request #7263 from planetscale/online-ddl-vtgate-session-uuid

Adding @@session_uuid to vtgate; used as 'context' by Online DDL

go/vt/sqlparser/ast_rewriting.go

@@ -234,6 +234,7 @@ func (er *expressionRewriter) sysVarRewrite(cursor *Cursor, node *ColName) {
 		sysvars.TransactionMode.Name,
 		sysvars.Workload.Name,
 		sysvars.DDLStrategy.Name,
+		sysvars.SessionUUID.Name,
 		sysvars.ReadAfterWriteGTID.Name,
 		sysvars.ReadAfterWriteTimeOut.Name,
 		sysvars.SessionTrackGTIDs.Name:

go/vt/sqlparser/ast_rewriting_test.go

@@ -29,7 +29,7 @@ import (
 type myTestCase struct {
 	in, expected                                                      string
 	liid, db, foundRows, rowCount, rawGTID, rawTimeout, sessTrackGTID bool
-	ddlStrategy                                                       bool
+	ddlStrategy, sessionUUID                                          bool
 	udv                                                               int
 	autocommit, clientFoundRows, skipQueryPlanCache                   bool
 	sqlSelectLimit, transactionMode, workload                         bool
@@ -194,6 +194,7 @@ func TestRewrites(in *testing.T) {
 			assert.Equal(tc.transactionMode, result.NeedsSysVar(sysvars.TransactionMode.Name), "should need :__vttransactionMode")
 			assert.Equal(tc.workload, result.NeedsSysVar(sysvars.Workload.Name), "should need :__vtworkload")
 			assert.Equal(tc.ddlStrategy, result.NeedsSysVar(sysvars.DDLStrategy.Name), "should need ddlStrategy")
+			assert.Equal(tc.sessionUUID, result.NeedsSysVar(sysvars.SessionUUID.Name), "should need sessionUUID")
 			assert.Equal(tc.rawGTID, result.NeedsSysVar(sysvars.ReadAfterWriteGTID.Name), "should need rawGTID")
 			assert.Equal(tc.rawTimeout, result.NeedsSysVar(sysvars.ReadAfterWriteTimeOut.Name), "should need rawTimeout")
 			assert.Equal(tc.sessTrackGTID, result.NeedsSysVar(sysvars.SessionTrackGTIDs.Name), "should need sessTrackGTID")

go/vt/sysvars/sysvars.go

@@ -53,6 +53,7 @@ var (
 	Workload            = SystemVariable{Name: "workload", IdentifierAsString: true}
 	Charset             = SystemVariable{Name: "charset", Default: utf8, IdentifierAsString: true}
 	Names               = SystemVariable{Name: "names", Default: utf8, IdentifierAsString: true}
+	SessionUUID         = SystemVariable{Name: "session_uuid", IdentifierAsString: true}
 	// Online DDL
 	DDLStrategy = SystemVariable{Name: "ddl_strategy", IdentifierAsString: true}
 
@@ -73,6 +74,7 @@ var (
 		Workload,
 		Charset,
 		Names,
+		SessionUUID,
 		ReadAfterWriteGTID,
 		ReadAfterWriteTimeOut,
 		SessionTrackGTIDs,

go/vt/vtgate/engine/fake_vcursor_test.go

@@ -65,6 +65,10 @@ func (t noopVCursor) GetDDLStrategy() string {
 	panic("implement me")
 }
 
+func (t noopVCursor) GetSessionUUID() string {
+	panic("implement me")
+}
+
 func (t noopVCursor) SetReadAfterWriteGTID(s string) {
 	panic("implement me")
 }

go/vt/vtgate/engine/online_ddl.go

@@ -17,6 +17,8 @@ limitations under the License.
 package engine
 
 import (
+	"fmt"
+
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/proto/query"
 	querypb "vitess.io/vitess/go/vt/proto/query"
@@ -75,7 +77,7 @@ func (v *OnlineDDL) Execute(vcursor VCursor, bindVars map[string]*query.BindVari
 	}
 	rows := [][]sqltypes.Value{}
 	for _, normalized := range normalizedQueries {
-		onlineDDL, err := schema.NewOnlineDDL(v.GetKeyspaceName(), normalized.TableName.Name.String(), normalized.SQL, v.Strategy, v.Options, "vtgate")
+		onlineDDL, err := schema.NewOnlineDDL(v.GetKeyspaceName(), normalized.TableName.Name.String(), normalized.SQL, v.Strategy, v.Options, fmt.Sprintf("vtgate:%s", vcursor.Session().GetSessionUUID()))
 		if err != nil {
 			return result, err
 		}

go/vt/vtgate/engine/primitive.go

@@ -131,6 +131,8 @@ type (
 		SetDDLStrategy(string)
 		GetDDLStrategy() string
 
+		GetSessionUUID() string
+
 		// SetReadAfterWriteGTID sets the GTID that the user expects a replica to have caught up with before answering a query
 		SetReadAfterWriteGTID(string)
 		SetReadAfterWriteTimeout(float64)

go/vt/vtgate/executor.go

@@ -294,6 +294,8 @@ func (e *Executor) addNeededBindVars(bindVarNeeds *sqlparser.BindVarNeeds, bindV
 			bindVars[key] = sqltypes.StringBindVariable(v)
 		case sysvars.DDLStrategy.Name:
 			bindVars[key] = sqltypes.StringBindVariable(session.DDLStrategy)
+		case sysvars.SessionUUID.Name:
+			bindVars[key] = sqltypes.StringBindVariable(session.SessionUUID)
 		case sysvars.ReadAfterWriteGTID.Name:
 			var v string
 			ifReadAfterWriteExist(session, func(raw *vtgatepb.ReadAfterWrite) {

go/vt/vtgate/plugin_mysql_server.go

@@ -34,10 +34,9 @@ import (
 
 	"context"
 
-	"vitess.io/vitess/go/trace"
-
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/sqltypes"
+	"vitess.io/vitess/go/trace"
 	"vitess.io/vitess/go/vt/callerid"
 	"vitess.io/vitess/go/vt/callinfo"
 	"vitess.io/vitess/go/vt/log"
@@ -46,6 +45,8 @@ import (
 
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	vtgatepb "vitess.io/vitess/go/vt/proto/vtgate"
+
+	"github.com/google/uuid"
 )
 
 var (
@@ -335,13 +336,15 @@ func (vh *vtgateHandler) WarningCount(c *mysql.Conn) uint16 {
 func (vh *vtgateHandler) session(c *mysql.Conn) *vtgatepb.Session {
 	session, _ := c.ClientData.(*vtgatepb.Session)
 	if session == nil {
+		u, _ := uuid.NewUUID()
 		session = &vtgatepb.Session{
 			Options: &querypb.ExecuteOptions{
 				IncludedFields: querypb.ExecuteOptions_ALL,
 				Workload:       querypb.ExecuteOptions_Workload(mysqlDefaultWorkload),
 			},
 			Autocommit:  true,
 			DDLStrategy: *defaultDDLStrategy,
+			SessionUUID: u.String(),
 		}
 		if c.Capabilities&mysql.CapabilityClientFoundRows != 0 {
 			session.Options.ClientFoundRows = true

go/vt/vtgate/safe_session.go

@@ -471,6 +471,13 @@ func (session *SafeSession) GetDDLStrategy() string {
 	return session.DDLStrategy
 }
 
+// GetSessionUUID returns the SessionUUID value.
+func (session *SafeSession) GetSessionUUID() string {
+	session.mu.Lock()
+	defer session.mu.Unlock()
+	return session.SessionUUID
+}
+
 // SetReadAfterWriteGTID set the ReadAfterWriteGtid setting.
 func (session *SafeSession) SetReadAfterWriteGTID(vtgtid string) {
 	session.mu.Lock()

go/vt/vtgate/vcursor_impl.go

@@ -619,6 +619,11 @@ func (vc *vcursorImpl) GetDDLStrategy() string {
 	return vc.safeSession.GetDDLStrategy()
 }
 
+// GetSessionUUID implements the SessionActions interface
+func (vc *vcursorImpl) GetSessionUUID() string {
+	return vc.safeSession.GetSessionUUID()
+}
+
 // SetReadAfterWriteGTID implements the SessionActions interface
 func (vc *vcursorImpl) SetReadAfterWriteGTID(vtgtid string) {
 	vc.safeSession.SetReadAfterWriteGTID(vtgtid)

proto/vtgate.proto

@@ -140,6 +140,9 @@ message Session {
 
   // DDL strategy
   string DDLStrategy = 21;
+
+  // Session UUID
+  string SessionUUID = 22;
 }
 
 // ReadAfterWrite contains information regarding gtid set and timeout