sql: add identifiers to sampled query

Partially addresses: #71328

This change introduces identifiers into the sampled query log, namely:
- Database name
- Session ID
- Transaction ID
- Statement ID

Note: Adding transaction ID incurs an additional lock access.

Release note (sql change): Sampled query telemetry log now includes
session/transaction/statement IDs, and database name of the query.

Release justification: Category 4: Low risk, high benefit changes to
existing functionality

docs/generated/eventlog.md

@@ -2242,6 +2242,10 @@ contains common SQL event/execution details.
 | `SkippedQueries` | skipped_queries indicate how many SQL statements were not considered for sampling prior to this one. If the field is omitted, or its value is zero, this indicates that no statement was omitted since the last event. | no |
 | `CostEstimate` | Cost of the query as estimated by the optimizer. | no |
 | `Distribution` | The distribution of the DistSQL query plan (local, full, or partial). | no |
+| `SessionID` | SessionID is the ID of the session that initiated the query. | no |
+| `Database` | Name of the database that initiated the query. | no |
+| `StatementID` | Statement ID of the query. | no |
+| `TransactionID` | Transaction ID of the query. | no |
 
 
 #### Common fields

pkg/sql/exec_log.go

@@ -382,6 +382,10 @@ func (p *planner) maybeLogStatementInternal(
 				SkippedQueries:       skippedQueries,
 				CostEstimate:         p.curPlan.instrumentation.costEstimate,
 				Distribution:         p.curPlan.instrumentation.distribution.String(),
+				SessionID:            p.extendedEvalCtx.SessionID.String(),
+				Database:             p.CurrentDatabase(),
+				StatementID:          p.stmt.QueryID.String(),
+				TransactionID:        p.txn.ID().String(),
 			}})
 		} else {
 			telemetryMetrics.incSkippedQueryCount()

pkg/sql/telemetry_logging_test.go

@@ -92,7 +92,12 @@ func TestTelemetryLogging(t *testing.T) {
 
 	defer s.Stopper().Stop(context.Background())
 
+	var sessionID string
+	var databaseName string
+
 	db := sqlutils.MakeSQLRunner(sqlDB)
+	db.QueryRow(t, `SHOW session_id`).Scan(&sessionID)
+	db.QueryRow(t, `SHOW database`).Scan(&databaseName)
 	db.Exec(t, `SET application_name = 'telemetry-logging-test'`)
 	db.Exec(t, `SET CLUSTER SETTING sql.telemetry.query_sampling.enabled = true;`)
 	db.Exec(t, "CREATE TABLE t();")
@@ -231,6 +236,16 @@ func TestTelemetryLogging(t *testing.T) {
 				if !distRe.MatchString(e.Message) {
 					t.Errorf("expected to find Distribution but none was found")
 				}
+				// Match StatementID on any non-empty string value.
+				stmtID := regexp.MustCompile("\"StatementID\":(\"\\S+\")")
+				if !stmtID.MatchString(e.Message) {
+					t.Errorf("expected to find StatementID but none was found in: %s", e.Message)
+				}
+				// Match TransactionID on any non-empty string value.
+				txnID := regexp.MustCompile("\"TransactionID\":(\"\\S+\")")
+				if !txnID.MatchString(e.Message) {
+					t.Errorf("expected to find TransactionID but none was found in: %s", e.Message)
+				}
 				for _, eTag := range tc.expectedUnredactedTags {
 					for _, tag := range strings.Split(e.Tags, ",") {
 						kv := strings.Split(tag, "=")
@@ -242,6 +257,12 @@ func TestTelemetryLogging(t *testing.T) {
 				if !strings.Contains(e.Message, "\"ApplicationName\":\""+tc.expectedApplicationName+"\"") {
 					t.Errorf("expected to find unredacted Application Name: %s", tc.expectedApplicationName)
 				}
+				if !strings.Contains(e.Message, "\"SessionID\":\""+sessionID+"\"") {
+					t.Errorf("expected to find sessionID: %s", sessionID)
+				}
+				if !strings.Contains(e.Message, "\"Database\":\""+databaseName+"\"") {
+					t.Errorf("expected to find Database: %s", databaseName)
+				}
 			}
 		}
 		if logCount != expectedLogCount {