sql,log: add new txn_timestamp filed to CommonSqlEventDetails

Previously, an event's `timestamp` field was either set to
the current time, or the read timestamp of the txn emitting
the event, if the event was emitted within a txn. This isn't
obvious functionality and can result in events to be seemingly
out of order.

To fix this, we adeded a new  `txn_timestamp` field to the
CommonSQLEventDetails  proto type. This field is responsible
for holding the txn's current read timestamp that a structured
event was emitted in, if in a transaction. Now, an event's
timestamp will always be the time that the event was
constructred, and txn_timestamp will transactions current
read timestamp.

Resolves: #146820
Epic: None
Release note (general change): Updated SQL events to have a
'TxnTimestamp' which indicates the read timestamp of the
transaction that the sql event was emitted in, if in
a transaction. The `Timestamp` of an event will always be
the time that the event was created.

Author: kyle-a-wong

docs/generated/eventlog.md

@@ -1097,7 +1097,7 @@ ALTER DATABASE drop_region_db DROP REGION IF EXISTS "us-east-1"
 
 # Ensure that events are generated for dropping the region
 query IT
-SELECT "reportingID", info::JSONB - 'Timestamp' - 'DescriptorID'
+SELECT "reportingID", info::JSONB - 'Timestamp' - 'DescriptorID' - 'TxnTimestamp'
 FROM system.eventlog
 WHERE "eventType" = 'alter_database_drop_region'
 ----

pkg/ccl/logictestccl/testdata/logic_test/multi_region

@@ -247,12 +247,8 @@ func logEventInternalForSQLStatements(
 ) error {
 	// Inject the common fields into the payload provided by the caller.
 	injectCommonFields := func(event logpb.EventPayload) error {
-		if txn == nil {
-			// No txn is set (e.g. for COPY or BEGIN), so use now instead.
-			event.CommonDetails().Timestamp = timeutil.Now().UnixNano()
-		} else {
-			event.CommonDetails().Timestamp = txn.KV().ReadTimestamp().WallTime
-		}
+		event.CommonDetails().Timestamp = timeutil.Now().UnixNano()
+
 		sqlCommon, ok := event.(eventpb.EventWithCommonSQLPayload)
 		if !ok {
 			return errors.AssertionFailedf("unknown event type: %T", event)
@@ -278,6 +274,10 @@ func logEventInternalForSQLStatements(
 		// Overwrite with the common details.
 		*m = commonSQLEventDetails
 
+		if txn != nil {
+			m.TxnTimestamp = txn.KV().ReadTimestamp().WallTime
+		}
+
 		// If the common details didn't have a descriptor ID, keep the
 		// one that was in the event already.
 		if m.DescriptorID == 0 {

pkg/sql/event_log.go

@@ -24,6 +24,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log/eventpb"
 	"github.com/cockroachdb/cockroach/pkg/util/log/logconfig"
 	"github.com/cockroachdb/cockroach/pkg/util/log/logpb"
+	"github.com/cockroachdb/cockroach/pkg/util/log/logtestutils"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
@@ -115,6 +116,46 @@ func TestStructuredEventLogging(t *testing.T) {
 	}
 }
 
+func TestStructuredEventLogging_txnTimestamp(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	appLogsSpy := logtestutils.NewStructuredLogSpy(
+		t,
+		[]logpb.Channel{logpb.Channel_SQL_SCHEMA},
+		[]string{"create_table"},
+		func(entry logpb.Entry) (eventpb.CreateTable, error) {
+			var cte eventpb.CreateTable
+			if err := json.Unmarshal([]byte(entry.Message[entry.StructuredStart:entry.StructuredEnd]), &cte); err != nil {
+				return cte, err
+			}
+			return cte, nil
+		},
+	)
+
+	cleanup := log.InterceptWith(ctx, appLogsSpy)
+	defer cleanup()
+
+	s, conn, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	defer s.Stopper().Stop(ctx)
+
+	runner := sqlutils.MakeSQLRunner(conn)
+	runner.Exec(t, "CREATE TABLE test (id INT PRIMARY KEY)")
+	runner.Exec(t, "BEGIN")
+	runner.Exec(t, "SET autocommit_before_ddl = false")
+	runner.Exec(t, "CREATE TABLE test1 (id INT PRIMARY KEY)")
+	runner.Exec(t, "CREATE TABLE test2 (id INT PRIMARY KEY)")
+	runner.Exec(t, "COMMIT")
+
+	createTables := appLogsSpy.GetLogs(logpb.Channel_SQL_SCHEMA)
+	require.Len(t, createTables, 3)
+	// Not created in the same transaction, so transaction timestamps are different
+	require.NotEqual(t, createTables[0].TxnTimestamp, createTables[1].TxnTimestamp)
+	// Created in the same transaction, so transaction timestamps are the same
+	require.Equal(t, createTables[1].TxnTimestamp, createTables[2].TxnTimestamp)
+}
+
 var execLogRe = regexp.MustCompile(`event_log.go`)
 
 // Test the SQL_PERF and SQL_INTERNAL_PERF logging channels.
@@ -140,7 +181,7 @@ func TestPerfLogging(t *testing.T) {
 		{
 			query:       `SELECT pg_sleep(0.256)`,
 			errRe:       ``,
-			logRe:       `"EventType":"slow_query","Statement":"SELECT pg_sleep\(‹0.256›\)","Tag":"SELECT","User":"root","ExecMode":"exec","NumRows":1`,
+			logRe:       `"EventType":"slow_query","Statement":"SELECT pg_sleep\(‹0.256›\)","Tag":"SELECT","User":"root","TxnTimestamp":.*,"ExecMode":"exec","NumRows":1`,
 			logExpected: true,
 			channel:     channel.SQL_PERF,
 		},
@@ -161,7 +202,7 @@ func TestPerfLogging(t *testing.T) {
 		{
 			query:       `INSERT INTO t VALUES (2, pg_sleep(0.256), 'x')`,
 			errRe:       ``,
-			logRe:       `"EventType":"slow_query","Statement":"INSERT INTO .*\.t VALUES \(‹2›, pg_sleep\(‹0.256›\), ‹'x'›\)","Tag":"INSERT","User":"root","ExecMode":"exec","NumRows":1`,
+			logRe:       `"EventType":"slow_query","Statement":"INSERT INTO .*\.t VALUES \(‹2›, pg_sleep\(‹0.256›\), ‹'x'›\)","Tag":"INSERT","User":"root","TxnTimestamp":.*,"ExecMode":"exec","NumRows":1`,
 			logExpected: true,
 			channel:     channel.SQL_PERF,
 		},
@@ -175,7 +216,7 @@ func TestPerfLogging(t *testing.T) {
 		{
 			query:       `INSERT INTO t VALUES (4, pg_sleep(0.256), repeat('x', 1024))`,
 			errRe:       ``,
-			logRe:       `"EventType":"slow_query","Statement":"INSERT INTO .*\.t VALUES \(‹4›, pg_sleep\(‹0.256›\), repeat\(‹'x'›, ‹1024›\)\)","Tag":"INSERT","User":"root","ExecMode":"exec","NumRows":1`,
+			logRe:       `"EventType":"slow_query","Statement":"INSERT INTO .*\.t VALUES \(‹4›, pg_sleep\(‹0.256›\), repeat\(‹'x'›, ‹1024›\)\)","Tag":"INSERT","User":"root","TxnTimestamp":.*,"ExecMode":"exec","NumRows":1`,
 			logExpected: true,
 			channel:     channel.SQL_PERF,
 		},
@@ -189,7 +230,7 @@ func TestPerfLogging(t *testing.T) {
 		{
 			query:       `SELECT *, pg_sleep(0.064) FROM t`,
 			errRe:       ``,
-			logRe:       `"EventType":"slow_query","Statement":"SELECT \*, pg_sleep\(‹0.064›\) FROM .*\.t","Tag":"SELECT","User":"root","ExecMode":"exec","NumRows":4`,
+			logRe:       `"EventType":"slow_query","Statement":"SELECT \*, pg_sleep\(‹0.064›\) FROM .*\.t","Tag":"SELECT","User":"root","TxnTimestamp":.*,"ExecMode":"exec","NumRows":4`,
 			logExpected: true,
 			channel:     channel.SQL_PERF,
 		},

pkg/sql/event_log_test.go

@@ -25,7 +25,7 @@ CREATE STATISTICS __auto__ FROM a
 # Check explicitly for table id 106. System tables could trigger autostats
 # collections at any time.
 query IT
-SELECT "reportingID", "info"::JSONB - 'Timestamp' - 'DescriptorID' - 'ApplicationName'
+SELECT "reportingID", "info"::JSONB - 'Timestamp' - 'DescriptorID' - 'TxnTimestamp' - 'ApplicationName'
 FROM system.eventlog
 WHERE "eventType" = 'create_statistics' AND ("info"::JSONB ->> 'DescriptorID')::INT = 106
 ORDER BY "timestamp", info