stmtdiagnostics: write transaction diagnostics to new system tables

Updates `TxnRegistry.InsertTxnRequest` and `TxnRegistry.InsertTxnDiagnostic`
to persist data to the new `system.transaction_diagnostics_requests` and
`system.transaction_diagnostics` tables.

`InsertTxnDiagnostic` will also update the `system.statement_diagnostics`
table to update the `transaction_diagnostics_id` column and update
the `system.transaction_diagnostics_requests` table to mark the request
as complete. It will also insert its own bundle into
`system.statement_bundle_chunks`. For now, this bundle will only contain
trace recordings, but in the future it can be updated to have any data
related to the transaction.

Part-of: CRDB-53546
Epic: CRDB-53541
Release note: None

Author: kyle-a-wong

pkg/sql/stmtdiagnostics/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
         "//pkg/sql/isql",
         "//pkg/sql/sem/tree",
         "//pkg/sql/sessiondata",
+        "//pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil",
         "//pkg/sql/types",
         "//pkg/util/intsets",
         "//pkg/util/log",
@@ -32,6 +33,7 @@ go_test(
         "main_test.go",
         "statement_diagnostics_helpers_test.go",
         "statement_diagnostics_test.go",
+        "txn_diagnostics_helpers_test.go",
         "txn_diagnostics_test.go",
     ],
     embed = [":stmtdiagnostics"],
@@ -56,12 +58,14 @@ go_test(
         "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",
+        "//pkg/util/encoding",
         "//pkg/util/leaktest",
         "//pkg/util/log",
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",
         "//pkg/util/uuid",
         "@com_github_cockroachdb_errors//:errors",
+        "@com_github_lib_pq//:pq",
         "@com_github_stretchr_testify//require",
     ],
 )

pkg/sql/stmtdiagnostics/statement_diagnostics.go

@@ -580,7 +580,7 @@ func (r *Registry) InsertStatementDiagnostics(
 	var diagID CollectedInstanceID
 	err := r.db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 		txn.KV().SetDebugName("stmt-diag-insert-bundle")
-		id, err := r.innerInsertStatementDiagnostics(ctx, NewStmtDiagnostic(requestID, req, stmtFingerprint, stmt, bundle, collectionErr), txn)
+		id, err := r.innerInsertStatementDiagnostics(ctx, NewStmtDiagnostic(requestID, req, stmtFingerprint, stmt, bundle, collectionErr), txn, CollectedInstanceID(0))
 		if err != nil {
 			return err
 		}
@@ -627,7 +627,7 @@ func (r *Registry) insertBundleChunks(
 }
 
 func (r *Registry) innerInsertStatementDiagnostics(
-	ctx context.Context, diagnostic StmtDiagnostic, txn isql.Txn,
+	ctx context.Context, diagnostic StmtDiagnostic, txn isql.Txn, txnDiagnosticId CollectedInstanceID,
 ) (CollectedInstanceID, error) {
 	var diagID CollectedInstanceID
 	if diagnostic.requestID != 0 {
@@ -663,14 +663,22 @@ func (r *Registry) innerInsertStatementDiagnostics(
 
 	collectionTime := timeutil.Now()
 
+	insertCols := "statement_fingerprint, statement, collected_at, bundle_chunks, error"
+	insertVals := "$1, $2, $3, $4, $5"
+	vals := []interface{}{diagnostic.stmtFingerprint, diagnostic.stmt, collectionTime, bundleChunksVal, errorVal}
+	if txnDiagnosticId != 0 {
+		insertCols += ", transaction_diagnostics_id"
+		insertVals += ", $6"
+		vals = append(vals, txnDiagnosticId)
+	}
 	// Insert the collection metadata into system.statement_diagnostics.
 	row, err := txn.QueryRowEx(
 		ctx, "stmt-diag-insert", txn.KV(),
 		sessiondata.NodeUserSessionDataOverride,
 		"INSERT INTO system.statement_diagnostics "+
-			"(statement_fingerprint, statement, collected_at, bundle_chunks, error) "+
-			"VALUES ($1, $2, $3, $4, $5) RETURNING id",
-		diagnostic.stmtFingerprint, diagnostic.stmt, collectionTime, bundleChunksVal, errorVal,
+			"("+insertCols+") "+
+			"VALUES ("+insertVals+") RETURNING id",
+		vals...,
 	)
 	if err != nil {
 		return diagID, err