insights: add status field to transaction insights

Part of: cockroachdb#87785.

Previously, execution insights did not keep track of a transaction's
status. This commit adds the `status` field to transaction insights. A
transaction's status is determined from its statements: if all
statement statuses are `Completed`, then the transaction status is
`Completed` and if there is one `Failed` statement status, then the
transaction status is also `Failed`.

Release note (sql change): Added `status` column to
`crdb_internal.[cluster|node]_txn_execution_insights` virtual tables.

pkg/ccl/logictestccl/testdata/logic_test/crdb_internal_tenant

@@ -323,15 +323,15 @@ SELECT * FROM crdb_internal.node_execution_insights WHERE query = ''
 ----
 session_id  txn_id  txn_fingerprint_id  stmt_id  stmt_fingerprint_id  problem  causes  query  status  start_time  end_time  full_scan  user_name  app_name  database_name  plan_gist  rows_read  rows_written  priority  retries  last_retry_reason  exec_node_ids  contention  index_recommendations  implicit_txn  cpu_sql_nanos  error_code
 
-query TTTBTTTTTIITITTTTTIT colnames
+query TTTBTTTTTIITITTTTTITT colnames
 SELECT * FROM crdb_internal.cluster_txn_execution_insights WHERE query = ''
 ----
-txn_id  txn_fingerprint_id  query  implicit_txn  session_id  start_time  end_time  user_name  app_name  rows_read  rows_written  priority  retries  last_retry_reason  contention  problems  causes  stmt_execution_ids  cpu_sql_nanos  last_error_code
+txn_id  txn_fingerprint_id  query  implicit_txn  session_id  start_time  end_time  user_name  app_name  rows_read  rows_written  priority  retries  last_retry_reason  contention  problems  causes  stmt_execution_ids  cpu_sql_nanos  last_error_code status
 
-query TTTBTTTTTIITITTTTTIT colnames
+query TTTBTTTTTIITITTTTTITT colnames
 SELECT * FROM crdb_internal.node_txn_execution_insights WHERE query = ''
 ----
-txn_id  txn_fingerprint_id  query  implicit_txn  session_id  start_time  end_time  user_name  app_name  rows_read  rows_written  priority  retries  last_retry_reason  contention  problems  causes  stmt_execution_ids  cpu_sql_nanos  last_error_code
+txn_id  txn_fingerprint_id  query  implicit_txn  session_id  start_time  end_time  user_name  app_name  rows_read  rows_written  priority  retries  last_retry_reason  contention  problems  causes  stmt_execution_ids  cpu_sql_nanos  last_error_code status
 
 query ITTI
 SELECT range_id, start_pretty, end_pretty, lease_holder FROM crdb_internal.ranges

pkg/sql/crdb_internal.go

@@ -7167,7 +7167,8 @@ CREATE TABLE crdb_internal.%s (
   causes                     STRING[] NOT NULL,
   stmt_execution_ids         STRING[] NOT NULL,
   cpu_sql_nanos              INT8,
-  last_error_code            STRING
+  last_error_code            STRING,
+  status                     STRING NOT NULL
 )`
 
 var crdbInternalClusterTxnExecutionInsightsTable = virtualSchemaTable{
@@ -7296,6 +7297,7 @@ func populateTxnExecutionInsights(
 			stmtIDs,
 			tree.NewDInt(tree.DInt(insight.Transaction.CPUSQLNanos)),
 			tree.NewDString(errorCode),
+			tree.NewDString(insight.Transaction.Status.String()),
 		))
 
 		if err != nil {

pkg/sql/logictest/testdata/logic_test/crdb_internal

@@ -463,16 +463,16 @@ SELECT * FROM crdb_internal.ranges_no_leases WHERE range_id < 0
 ----
 range_id  start_key  start_pretty  end_key  end_pretty  replicas  replica_localities  voting_replicas  non_voting_replicas  learner_replicas  split_enforced_until
 
-query TTTBTTTTTIITITTTTTIT colnames
+query TTTBTTTTTIITITTTTTITT colnames
 SELECT * FROM crdb_internal.cluster_txn_execution_insights WHERE query = ''
 ----
-txn_id  txn_fingerprint_id  query  implicit_txn  session_id  start_time  end_time  user_name  app_name  rows_read  rows_written  priority  retries  last_retry_reason  contention  problems  causes  stmt_execution_ids cpu_sql_nanos last_error_code
+txn_id  txn_fingerprint_id  query  implicit_txn  session_id  start_time  end_time  user_name  app_name  rows_read  rows_written  priority  retries  last_retry_reason  contention  problems  causes  stmt_execution_ids cpu_sql_nanos last_error_code status
 
 
-query TTTBTTTTTIITITTTTTIT colnames
+query TTTBTTTTTIITITTTTTITT colnames
 SELECT * FROM crdb_internal.node_txn_execution_insights WHERE query = ''
 ----
-txn_id  txn_fingerprint_id  query  implicit_txn  session_id  start_time  end_time  user_name  app_name  rows_read  rows_written  priority  retries  last_retry_reason  contention  problems  causes  stmt_execution_ids cpu_sql_nanos last_error_code
+txn_id  txn_fingerprint_id  query  implicit_txn  session_id  start_time  end_time  user_name  app_name  rows_read  rows_written  priority  retries  last_retry_reason  contention  problems  causes  stmt_execution_ids cpu_sql_nanos last_error_code status
 
 
 statement ok

pkg/sql/sqlstats/insights/insights.proto

@@ -60,6 +60,12 @@ message Session {
 }
 
 message Transaction {
+  // TODO(gtr): Implement "Cancelled" status for transaction insights.
+  enum Status {
+    Completed = 0;
+    Failed = 1;
+  }
+
   bytes id = 1 [(gogoproto.customname) = "ID",
     (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/uuid.UUID",
     (gogoproto.nullable) = false];
@@ -85,6 +91,7 @@ message Transaction {
   int64 cpu_sql_nanos = 17 [(gogoproto.customname) = "CPUSQLNanos"];
   // The error code of the last failed statement in the transaction.
   string last_error_code = 18; 
+  Status status = 19;
 }
 
 message Statement {

pkg/sql/sqlstats/insights/integration/insights_test.go

@@ -286,6 +286,7 @@ func TestFailedInsights(t *testing.T) {
 		problems    string
 		errorCode   string
 		endTxn      bool
+		txnStatus   string
 	}{
 		{
 			// Single-statement txn that will fail.
@@ -294,6 +295,7 @@ func TestFailedInsights(t *testing.T) {
 			problems:    "{FailedExecution}",
 			errorCode:   "42501",
 			endTxn:      true,
+			txnStatus:   "Failed",
 		},
 		{
 			// Multi-statement txn that will fail.
@@ -302,6 +304,7 @@ func TestFailedInsights(t *testing.T) {
 			problems:    "{FailedExecution}",
 			errorCode:   "22012",
 			endTxn:      true,
+			txnStatus:   "Failed",
 		},
 		{
 			// Multi-statement txn with a slow stmt and then a failed execution.
@@ -310,6 +313,7 @@ func TestFailedInsights(t *testing.T) {
 			problems:    "{SlowExecution,FailedExecution}",
 			errorCode:   "42P07",
 			endTxn:      true,
+			txnStatus:   "Failed",
 		},
 		{
 			// Multi-statement txn with a slow stmt but no failures.
@@ -318,6 +322,7 @@ func TestFailedInsights(t *testing.T) {
 			problems:    "{SlowExecution}",
 			errorCode:   "",
 			endTxn:      false,
+			txnStatus:   "Completed",
 		},
 	}
 
@@ -329,17 +334,18 @@ func TestFailedInsights(t *testing.T) {
 
 		testutils.SucceedsWithin(t, func() error {
 			var row *gosql.Row
-			var query, problems, errorCode string
+			var query, problems, status, errorCode string
 
 			// Query the node txn execution insights table.
 			row = conn.QueryRowContext(ctx, "SELECT "+
 				"query, "+
 				"problems, "+
+				"status, "+
 				"COALESCE(last_error_code, '') last_error_code "+
 				"FROM crdb_internal.node_txn_execution_insights "+
 				"WHERE query = $1 AND app_name = $2 ", tc.fingerprint, appName)
 
-			err = row.Scan(&query, &problems, &errorCode)
+			err = row.Scan(&query, &problems, &status, &errorCode)
 
 			if err != nil {
 				return err
@@ -349,6 +355,10 @@ func TestFailedInsights(t *testing.T) {
 				return fmt.Errorf("expected problems to be '%s', but was '%s'. stmts: %s", tc.problems, problems, tc.stmts)
 			}
 
+			if status != tc.txnStatus {
+				return fmt.Errorf("expected status to be '%s', but was '%s'. stmts: %s", tc.txnStatus, status, tc.stmts)
+			}
+
 			if errorCode != tc.errorCode {
 				return fmt.Errorf("expected error code to be '%s', but was '%s'. stmts: %s", tc.errorCode, errorCode, tc.stmts)
 			}

pkg/sql/sqlstats/insights/registry.go

@@ -128,6 +128,10 @@ func (r *lockingRegistry) ObserveTransaction(sessionID clusterunique.ID, transac
 	}
 
 	var lastErrorCode string
+	// The transaction status will reflect the status of its statements; it will
+	// default to completed unless a failed statement status is found. Note that
+	// this does not take into account the "Cancelled" transaction status.
+	var lastStatus = Transaction_Completed
 	for i, s := range *statements {
 		if slowOrFailedStatements.Contains(i) {
 			switch s.Status {
@@ -136,6 +140,7 @@ func (r *lockingRegistry) ObserveTransaction(sessionID clusterunique.ID, transac
 				s.Causes = r.causes.examine(s.Causes, s)
 			case Statement_Failed:
 				lastErrorCode = s.ErrorCode
+				lastStatus = Transaction_Status(s.Status)
 				s.Problem = Problem_FailedExecution
 			}
 
@@ -151,6 +156,7 @@ func (r *lockingRegistry) ObserveTransaction(sessionID clusterunique.ID, transac
 	}
 
 	insight.Transaction.LastErrorCode = lastErrorCode
+	insight.Transaction.Status = lastStatus
 	r.sink.AddInsight(insight)
 }
 

pkg/sql/sqlstats/insights/registry_test.go

@@ -76,6 +76,7 @@ func TestRegistry(t *testing.T) {
 		)
 
 		require.Equal(t, expected[0], actual[0])
+		require.Equal(t, transaction.Status, Transaction_Status(statement.Status))
 	})
 
 	t.Run("failure detection", func(t *testing.T) {
@@ -112,6 +113,7 @@ func TestRegistry(t *testing.T) {
 
 		require.Equal(t, expected, actual)
 		require.Equal(t, transaction.LastErrorCode, statement.ErrorCode)
+		require.Equal(t, transaction.Status, Transaction_Status(statement.Status))
 	})
 
 	t.Run("disabled", func(t *testing.T) {
@@ -253,6 +255,7 @@ func TestRegistry(t *testing.T) {
 		)
 
 		require.Equal(t, expected, actual)
+		require.Equal(t, transaction.Status, Transaction_Status(statement.Status))
 	})
 
 	t.Run("txn with no stmts", func(t *testing.T) {
@@ -301,5 +304,6 @@ func TestRegistry(t *testing.T) {
 		)
 
 		require.Equal(t, expected, actual)
+		require.Equal(t, transaction.Status, Transaction_Status(statement.Status))
 	})
 }