updated show plan type and transaction query plan types, added e2e te…

…st for metrics

Signed-off-by: Harshit Gangal <[email protected]>

go/test/endtoend/vtgate/misc_test.go

@@ -926,3 +926,73 @@ func getVar(t *testing.T, key string) interface{} {
 	}
 	return val
 }
+
+// TestQueryProcessedMetric tests that query processed metric is published.
+func TestQueryProcessedMetric(t *testing.T) {
+	conn, closer := start(t)
+	defer closer()
+
+	tcases := []struct {
+		sql    string
+		metric string
+	}{{
+		sql:    "select id1, id2 from t1",
+		metric: "SELECT.Scatter.PRIMARY",
+	}, {
+		sql:    "update t1 set id2 = 2 where id1 = 1",
+		metric: "UPDATE.Passthrough.PRIMARY",
+	}, {
+		sql:    "delete from t1 where id1 in (1, 2)",
+		metric: "DELETE.MultiShard.PRIMARY",
+	}, {
+		sql:    "show tables",
+		metric: "SHOW.Passthrough.PRIMARY",
+	}, {
+		sql:    "savepoint a",
+		metric: "SAVEPOINT.Transaction.PRIMARY",
+	}, {
+		sql:    "rollback",
+		metric: "ROLLBACK.Transaction.PRIMARY",
+	}}
+
+	initial := getQPMetric(t)
+	for _, tc := range tcases {
+		t.Run(tc.sql, func(t *testing.T) {
+			utils.Exec(t, conn, tc.sql)
+			updatedMetric := getQPMetric(t)
+			assert.EqualValues(t, 1, getValue(updatedMetric, tc.metric)-getValue(initial, tc.metric))
+		})
+	}
+}
+
+func getQPMetric(t *testing.T) map[string]any {
+	t.Helper()
+
+	vars := clusterInstance.VtgateProcess.GetVars()
+	require.NotNil(t, vars)
+
+	qpVars, exists := vars["QueryProcessed"]
+	if !exists {
+		return nil
+	}
+
+	qpMap, ok := qpVars.(map[string]any)
+	require.True(t, ok, "query processed vars is not a map")
+
+	return qpMap
+}
+
+func getValue(m map[string]any, key string) float64 {
+	if m == nil {
+		return 0
+	}
+	val, exists := m[key]
+	if !exists {
+		return 0
+	}
+	f, ok := val.(float64)
+	if !ok {
+		return 0
+	}
+	return f
+}

go/vt/vtgate/engine/plan.go

@@ -66,6 +66,7 @@ const (
 	PlanComplex
 	PlanOnlineDDL
 	PlanDirectDDL
+	PlanTransaction
 )
 
 func NewPlan(query string, stmt sqlparser.Statement, primitive Primitive, bindVarNeeds *sqlparser.BindVarNeeds, tablesUsed []string) *Plan {
@@ -146,6 +147,8 @@ func (p PlanType) String() string {
 		return "OnlineDDL"
 	case PlanDirectDDL:
 		return "DirectDDL"
+	case PlanTransaction:
+		return "Transaction"
 	default:
 		return "Unknown"
 	}
@@ -187,6 +190,8 @@ func getPlanType(p Primitive) PlanType {
 		return PlanDirectDDL
 	case *VExplain:
 		return getPlanType(prim.Input)
+	case *RenameFields:
+		return getPlanType(prim.Input)
 	default:
 		return PlanComplex
 	}

go/vt/vtgate/executor.go

@@ -592,8 +592,8 @@ func ifReadAfterWriteExist(session *econtext.SafeSession, f func(*vtgatepb.ReadA
 func (e *Executor) handleBegin(ctx context.Context, vcursor *econtext.VCursorImpl, safeSession *econtext.SafeSession, logStats *logstats.LogStats, stmt sqlparser.Statement) (*sqltypes.Result, error) {
 	execStart := time.Now()
 	logStats.PlanTime = execStart.Sub(logStats.StartTime)
-	e.updateQueryCounts("Begin", "", "", 0)
-	e.updateQueryStats("Begin", "Transaction", vcursor.TabletType().String())
+	e.updateQueryCounts(sqlparser.StmtBegin.String(), "", "", 0)
+	e.updateQueryStats(sqlparser.StmtBegin.String(), engine.PlanTransaction.String(), vcursor.TabletType().String())
 
 	begin := stmt.(*sqlparser.Begin)
 	err := e.txConn.Begin(ctx, safeSession, begin.TxAccessModes)
@@ -605,8 +605,8 @@ func (e *Executor) handleCommit(ctx context.Context, vcursor *econtext.VCursorIm
 	execStart := time.Now()
 	logStats.PlanTime = execStart.Sub(logStats.StartTime)
 	logStats.ShardQueries = uint64(len(safeSession.ShardSessions))
-	e.updateQueryCounts("Commit", "", "", int64(logStats.ShardQueries))
-	e.updateQueryStats("Commit", "Transaction", vcursor.TabletType().String())
+	e.updateQueryCounts(sqlparser.StmtCommit.String(), "", "", int64(logStats.ShardQueries))
+	e.updateQueryStats(sqlparser.StmtCommit.String(), engine.PlanTransaction.String(), vcursor.TabletType().String())
 
 	err := e.txConn.Commit(ctx, safeSession)
 	logStats.CommitTime = time.Since(execStart)
@@ -622,8 +622,8 @@ func (e *Executor) handleRollback(ctx context.Context, vcursor *econtext.VCursor
 	execStart := time.Now()
 	logStats.PlanTime = execStart.Sub(logStats.StartTime)
 	logStats.ShardQueries = uint64(len(safeSession.ShardSessions))
-	e.updateQueryCounts("Rollback", "", "", int64(logStats.ShardQueries))
-	e.updateQueryStats("Rollback", "Transaction", vcursor.TabletType().String())
+	e.updateQueryCounts(sqlparser.StmtRollback.String(), "", "", int64(logStats.ShardQueries))
+	e.updateQueryStats(sqlparser.StmtRollback.String(), engine.PlanTransaction.String(), vcursor.TabletType().String())
 
 	err := e.txConn.Rollback(ctx, safeSession)
 	logStats.CommitTime = time.Since(execStart)
@@ -635,7 +635,7 @@ func (e *Executor) handleSavepoint(ctx context.Context, vcursor *econtext.VCurso
 	logStats.PlanTime = execStart.Sub(logStats.StartTime)
 	logStats.ShardQueries = uint64(len(safeSession.ShardSessions))
 	e.updateQueryCounts(queryType, "", "", int64(logStats.ShardQueries))
-	e.updateQueryStats(queryType, "Transaction", vcursor.TabletType().String())
+	e.updateQueryStats(queryType, engine.PlanTransaction.String(), vcursor.TabletType().String())
 
 	defer func() {
 		logStats.ExecuteTime = time.Since(execStart)

go/vt/vtgate/plan_execute.go

@@ -245,19 +245,19 @@ func (e *Executor) handleTransactions(
 		qr, err := e.handleRollback(ctx, vcursor, safeSession, logStats)
 		return qr, err
 	case sqlparser.StmtSavepoint:
-		qr, err := e.handleSavepoint(ctx, vcursor, safeSession, plan.Original, "Savepoint", logStats, func(_ string) (*sqltypes.Result, error) {
+		qr, err := e.handleSavepoint(ctx, vcursor, safeSession, plan.Original, plan.QueryType.String(), logStats, func(_ string) (*sqltypes.Result, error) {
 			// Safely to ignore as there is no transaction.
 			return &sqltypes.Result{}, nil
 		}, vcursor.IgnoreMaxMemoryRows())
 		return qr, err
 	case sqlparser.StmtSRollback:
-		qr, err := e.handleSavepoint(ctx, vcursor, safeSession, plan.Original, "Rollback Savepoint", logStats, func(query string) (*sqltypes.Result, error) {
+		qr, err := e.handleSavepoint(ctx, vcursor, safeSession, plan.Original, plan.QueryType.String(), logStats, func(query string) (*sqltypes.Result, error) {
 			// Error as there is no transaction, so there is no savepoint that exists.
 			return nil, vterrors.NewErrorf(vtrpcpb.Code_NOT_FOUND, vterrors.SPDoesNotExist, "SAVEPOINT does not exist: %s", query)
 		}, vcursor.IgnoreMaxMemoryRows())
 		return qr, err
 	case sqlparser.StmtRelease:
-		qr, err := e.handleSavepoint(ctx, vcursor, safeSession, plan.Original, "Release Savepoint", logStats, func(query string) (*sqltypes.Result, error) {
+		qr, err := e.handleSavepoint(ctx, vcursor, safeSession, plan.Original, plan.QueryType.String(), logStats, func(query string) (*sqltypes.Result, error) {
 			// Error as there is no transaction, so there is no savepoint that exists.
 			return nil, vterrors.NewErrorf(vtrpcpb.Code_NOT_FOUND, vterrors.SPDoesNotExist, "SAVEPOINT does not exist: %s", query)
 		}, vcursor.IgnoreMaxMemoryRows())

go/vt/vtgate/planbuilder/testdata/show_cases.json

@@ -462,7 +462,7 @@
     "comment": "show tables",
     "query": "show tables",
     "plan": {
-      "Type": "Complex",
+      "Type": "Passthrough",
       "QueryType": "SHOW",
       "Original": "show tables",
       "Instructions": {
@@ -492,7 +492,7 @@
     "comment": "show tables from db",
     "query": "show tables from user",
     "plan": {
-      "Type": "Complex",
+      "Type": "Passthrough",
       "QueryType": "SHOW",
       "Original": "show tables from user",
       "Instructions": {
@@ -522,7 +522,7 @@
     "comment": "show tables from system schema",
     "query": "show tables from performance_schema",
     "plan": {
-      "Type": "Complex",
+      "Type": "Passthrough",
       "QueryType": "SHOW",
       "Original": "show tables from performance_schema",
       "Instructions": {