Merge cockroachdb#107081 cockroachdb#107437

107081: sql: add `StmtPosInTxn` to CommonSQLExecDetails r=xinhaoz a=xinhaoz

This commit adds the `StmtPosInTxn` field to the `CommonSQLExecDetails` event. `StmtPosInTxn` represents the stmt's index in the transaction, starting at 1.

Epic: none

Release note (sql change): In `CommonSQLExecDetails`, which is emitted as part of the sql audit logs, sql exec logs and telemetry events, there is a new field:
- `StmtPosInTxn`: represents the stmt's index in the transaction, starting at 1.

107437: testserver: reduce wait for nodelocal capability r=knz a=stevendanna

This adds the ability to restart the rangefeedcache used by the tenant capability watcher and then issues such a restart before waiting for the nodelocal capability.

The initial scan the restart forces typically takes less time than waiting out the closed timestamp.

Epic: CRDB-18499

Release note: None

Co-authored-by: Xin Hao Zhang <[email protected]>
Co-authored-by: Steven Danna <[email protected]>

Author: 3 people

docs/generated/eventlog.md

@@ -438,6 +438,7 @@ is directly or indirectly a member of the admin role) executes a query.
 | `FullIndexScan` | Whether the query contains a full secondary index scan of a non-partial index. | no |
 | `TxnCounter` | The sequence number of the SQL transaction inside its session. | no |
 | `BulkJobId` | The job id for bulk job (IMPORT/BACKUP/RESTORE). | no |
+| `StmtPosInTxn` | The statement's index in the transaction, starting at 1. | no |
 
 ### `role_based_audit_event`
 
@@ -473,6 +474,7 @@ cluster setting.
 | `FullIndexScan` | Whether the query contains a full secondary index scan of a non-partial index. | no |
 | `TxnCounter` | The sequence number of the SQL transaction inside its session. | no |
 | `BulkJobId` | The job id for bulk job (IMPORT/BACKUP/RESTORE). | no |
+| `StmtPosInTxn` | The statement's index in the transaction, starting at 1. | no |
 
 ### `sensitive_table_access`
 
@@ -508,6 +510,7 @@ a table marked as audited.
 | `FullIndexScan` | Whether the query contains a full secondary index scan of a non-partial index. | no |
 | `TxnCounter` | The sequence number of the SQL transaction inside its session. | no |
 | `BulkJobId` | The job id for bulk job (IMPORT/BACKUP/RESTORE). | no |
+| `StmtPosInTxn` | The statement's index in the transaction, starting at 1. | no |
 
 ## SQL Execution Log
 
@@ -550,6 +553,7 @@ and the cluster setting `sql.trace.log_statement_execute` is set.
 | `FullIndexScan` | Whether the query contains a full secondary index scan of a non-partial index. | no |
 | `TxnCounter` | The sequence number of the SQL transaction inside its session. | no |
 | `BulkJobId` | The job id for bulk job (IMPORT/BACKUP/RESTORE). | no |
+| `StmtPosInTxn` | The statement's index in the transaction, starting at 1. | no |
 
 ## SQL Logical Schema Changes
 
@@ -2350,6 +2354,7 @@ set to a non-zero value, AND
 | `FullIndexScan` | Whether the query contains a full secondary index scan of a non-partial index. | no |
 | `TxnCounter` | The sequence number of the SQL transaction inside its session. | no |
 | `BulkJobId` | The job id for bulk job (IMPORT/BACKUP/RESTORE). | no |
+| `StmtPosInTxn` | The statement's index in the transaction, starting at 1. | no |
 
 ### `txn_rows_read_limit`
 
@@ -2470,6 +2475,7 @@ the "slow query" condition.
 | `FullIndexScan` | Whether the query contains a full secondary index scan of a non-partial index. | no |
 | `TxnCounter` | The sequence number of the SQL transaction inside its session. | no |
 | `BulkJobId` | The job id for bulk job (IMPORT/BACKUP/RESTORE). | no |
+| `StmtPosInTxn` | The statement's index in the transaction, starting at 1. | no |
 
 ### `txn_rows_read_limit_internal`
 
@@ -3021,6 +3027,7 @@ contains common SQL event/execution details.
 | `FullIndexScan` | Whether the query contains a full secondary index scan of a non-partial index. | no |
 | `TxnCounter` | The sequence number of the SQL transaction inside its session. | no |
 | `BulkJobId` | The job id for bulk job (IMPORT/BACKUP/RESTORE). | no |
+| `StmtPosInTxn` | The statement's index in the transaction, starting at 1. | no |
 
 ### `schema_descriptor`
 

pkg/cli/interactive_tests/test_exec_log.tcl

@@ -116,12 +116,12 @@ grep 'SELECT ..*550..* +' $logfile;
 exit 1;"
 
 # Two separate single-stmt txns.
-system "n=`grep 'SELECT ..*550..* +' $logfile | sed -e 's/.*TxnCounter.:\\(\[0-9\]*\\)/\\1/g' | uniq | wc -l`; if test \$n -ne 2; then echo unexpected \$n; exit 1; fi"
+system "n=`grep 'SELECT ..*550..* +' $logfile | sed -e 's/.*TxnCounter.:\\(\[0-9\]*\\).*/\\1/g' | uniq | wc -l`; if test \$n -ne 2; then echo unexpected \$n; exit 1; fi"
 # Same txns.
-system "n=`grep 'SELECT ..*660..* +' $logfile | sed -e 's/.*TxnCounter.:\\(\[0-9\]*\\)/\\1/g' | uniq | wc -l`; if test \$n -ne 1; then echo unexpected \$n; exit 1; fi"
-system "n=`grep 'SELECT ..*770..* +' $logfile | sed -e 's/.*TxnCounter.:\\(\[0-9\]*\\)/\\1/g' | uniq | wc -l`; if test \$n -ne 1; then echo unexpected \$n; exit 1; fi"
-system "n=`grep 'SELECT ..*880..* +' $logfile | sed -e 's/.*TxnCounter.:\\(\[0-9\]*\\)/\\1/g' | uniq | wc -l`; if test \$n -ne 1; then echo unexpected \$n; exit 1; fi"
-system "n=`grep 'SELECT ..*990..* +' $logfile | sed -e 's/.*TxnCounter.:\\(\[0-9\]*\\)/\\1/g' | uniq | wc -l`; if test \$n -ne 1; then echo unexpected \$n; exit 1; fi"
+system "n=`grep 'SELECT ..*660..* +' $logfile | sed -e 's/.*TxnCounter.:\\(\[0-9\]*\\).*/\\1/g' | uniq | wc -l`; if test \$n -ne 1; then echo unexpected \$n; exit 1; fi"
+system "n=`grep 'SELECT ..*770..* +' $logfile | sed -e 's/.*TxnCounter.:\\(\[0-9\]*\\).*/\\1/g' | uniq | wc -l`; if test \$n -ne 1; then echo unexpected \$n; exit 1; fi"
+system "n=`grep 'SELECT ..*880..* +' $logfile | sed -e 's/.*TxnCounter.:\\(\[0-9\]*\\).*/\\1/g' | uniq | wc -l`; if test \$n -ne 1; then echo unexpected \$n; exit 1; fi"
+system "n=`grep 'SELECT ..*990..* +' $logfile | sed -e 's/.*TxnCounter.:\\(\[0-9\]*\\).*/\\1/g' | uniq | wc -l`; if test \$n -ne 1; then echo unexpected \$n; exit 1; fi"
 
 end_test
 

pkg/kv/kvclient/rangefeed/rangefeedcache/watcher.go

@@ -77,6 +77,9 @@ type Watcher struct {
 
 	lastFrontierTS hlc.Timestamp // used to assert monotonicity across rangefeed attempts
 
+	// Used to force a restart during testing.
+	restartErrCh chan error
+
 	knobs TestingKnobs
 }
 
@@ -177,6 +180,7 @@ func NewWatcher(
 		withPrevValue:    withPrevValue,
 		translateEvent:   translateEvent,
 		onUpdate:         onUpdate,
+		restartErrCh:     make(chan error),
 	}
 	if knobs != nil {
 		w.knobs = *knobs
@@ -353,12 +357,24 @@ func (s *Watcher) Run(ctx context.Context) error {
 
 		case err := <-errCh:
 			return err
+		case err := <-s.restartErrCh:
+			return err
 		case err := <-s.knobs.ErrorInjectionCh:
 			return err
 		}
 	}
 }
 
+var restartErr = errors.New("testing restart requested")
+
+// TestingRestart injects an error into the rangefeed cache, forcing
+// it to restart. This is separate from the testing knob so that we
+// can force a restart from test infrastructure without overriding the
+// user-provided testing knobs.
+func (s *Watcher) TestingRestart() {
+	s.restartErrCh <- restartErr
+}
+
 func (s *Watcher) handleUpdate(
 	ctx context.Context, buffer *rangefeedbuffer.Buffer, ts hlc.Timestamp, updateType UpdateType,
 ) {

pkg/multitenant/tenantcapabilities/tenantcapabilitieswatcher/watcher.go

@@ -54,6 +54,10 @@ type Watcher struct {
 	startCh  chan struct{}
 	startErr error
 
+	// rfc provides access to the underlying
+	// rangefeedcache.Watcher for testing.
+	rfc *rangefeedcache.Watcher
+
 	// initialScan is used to synchronize the Start() method with the
 	// reception of the initial batch of values from the rangefeed
 	// (which happens asynchronously).
@@ -212,6 +216,7 @@ func (w *Watcher) startRangeFeed(ctx context.Context) error {
 	if err := rangefeedcache.Start(ctx, w.stopper, rfc, w.onError); err != nil {
 		return err
 	}
+	w.rfc = rfc
 
 	// Wait for the initial scan before returning.
 	select {
@@ -346,6 +351,12 @@ func (w *Watcher) handleIncrementalUpdate(
 	}
 }
 
+func (w *Watcher) TestingRestart() {
+	if w.rfc != nil {
+		w.rfc.TestingRestart()
+	}
+}
+
 // TestingFlushCapabilitiesState flushes the underlying global tenant capability
 // state for testing purposes. The returned entries are sorted by tenant ID.
 func (w *Watcher) TestingFlushCapabilitiesState() (entries []tenantcapabilities.Entry) {

pkg/server/testserver.go

@@ -1201,18 +1201,13 @@ func (ts *TestServer) StartTenant(
 	baseCfg.DisableTLSForHTTP = params.DisableTLSForHTTP
 	baseCfg.EnableDemoLoginEndpoint = params.EnableDemoLoginEndpoint
 
-	// Waiting for capabilities can take 3+ seconds since the
-	// rangefeedcache needs to wait for the closed timestamp
-	// before flushing updates. To avoid paying this cost in all
-	// cases, we only set the nodelocal storage capability if the
-	// caller has configured an ExternalIODir since nodelocal
-	// storage only works with that configured.
+	// Waiting for capabilities can time To avoid paying this cost in all
+	// cases, we only set the nodelocal storage capability if the caller has
+	// configured an ExternalIODir since nodelocal storage only works with
+	// that configured.
 	//
-	// TODO(ssd): We should do more here. We could have the caller
-	// pass in explicitly that they want these capabilities. Or,
-	// we could modify the system in some way to avoid waiting on
-	// capabilities for so long. Also, note that this doesn't
-	// apply to StartSharedProcessTenant.
+	// TODO(ssd): We do not set this capability in
+	// StartSharedProcessTenant.
 	shouldGrantNodelocalCap := ts.params.ExternalIODir != ""
 	canGrantNodelocalCap := ts.ClusterSettings().Version.IsActive(ctx, clusterversion.V23_1TenantCapabilities)
 	if canGrantNodelocalCap && shouldGrantNodelocalCap {
@@ -1225,6 +1220,11 @@ func (ts *TestServer) StartTenant(
 				return nil, err
 			}
 		} else {
+			// Restart the capabilities watcher. Restarting the
+			// watcher forces a new initial scan which is faster
+			// than waiting out the closed timestamp interval
+			// required to see new updates.
+			ts.tenantCapabilitiesWatcher.TestingRestart()
 			if err := testutils.SucceedsSoonError(func() error {
 				capabilities, found := ts.TenantCapabilitiesReader().GetCapabilities(params.TenantID)
 				if !found {

pkg/sql/conn_executor.go

@@ -2792,6 +2792,7 @@ func (ex *connExecutor) execCopyOut(
 			int(ex.state.mu.autoRetryCounter),
 			ex.extraTxnState.txnCounter,
 			numOutputRows,
+			ex.state.mu.stmtCount,
 			0, /* bulkJobId */
 			copyErr,
 			ex.statsCollector.PhaseTimes().GetSessionPhaseTime(sessionphase.SessionQueryReceived),
@@ -3036,7 +3037,8 @@ func (ex *connExecutor) execCopyIn(
 		var stats topLevelQueryStats
 		ex.planner.maybeLogStatement(ctx, ex.executorType, true,
 			int(ex.state.mu.autoRetryCounter), ex.extraTxnState.txnCounter,
-			numInsertedRows, 0, /* bulkJobId */
+			numInsertedRows, ex.state.mu.stmtCount,
+			0, /* bulkJobId */
 			copyErr,
 			ex.statsCollector.PhaseTimes().GetSessionPhaseTime(sessionphase.SessionQueryReceived),
 			&ex.extraTxnState.hasAdminRoleCache,

pkg/sql/conn_executor_exec.go

@@ -1512,6 +1512,7 @@ func (ex *connExecutor) dispatchToExecutionEngine(
 						int(ex.state.mu.autoRetryCounter),
 						ex.extraTxnState.txnCounter,
 						ppInfo.dispatchToExecutionEngine.rowsAffected,
+						ex.state.mu.stmtCount,
 						bulkJobId,
 						ppInfo.curRes.ErrAllowReleased(),
 						ex.statsCollector.PhaseTimes().GetSessionPhaseTime(sessionphase.SessionQueryReceived),
@@ -1539,6 +1540,7 @@ func (ex *connExecutor) dispatchToExecutionEngine(
 				int(ex.state.mu.autoRetryCounter),
 				ex.extraTxnState.txnCounter,
 				nonBulkJobNumRows,
+				ex.state.mu.stmtCount,
 				bulkJobId,
 				res.Err(),
 				ex.statsCollector.PhaseTimes().GetSessionPhaseTime(sessionphase.SessionQueryReceived),

pkg/sql/exec_log.go

@@ -121,7 +121,7 @@ func (p *planner) maybeLogStatement(
 	ctx context.Context,
 	execType executorType,
 	isCopy bool,
-	numRetries, txnCounter, rows int,
+	numRetries, txnCounter, rows, stmtCount int,
 	bulkJobId uint64,
 	err error,
 	queryReceived time.Time,
@@ -133,7 +133,7 @@ func (p *planner) maybeLogStatement(
 ) {
 	p.maybeAuditRoleBasedAuditEvent(ctx, execType)
 	p.maybeLogStatementInternal(ctx, execType, isCopy, numRetries, txnCounter,
-		rows, bulkJobId, err, queryReceived, hasAdminRoleCache,
+		rows, stmtCount, bulkJobId, err, queryReceived, hasAdminRoleCache,
 		telemetryLoggingMetrics, stmtFingerprintID, queryStats, statsCollector,
 	)
 }
@@ -142,7 +142,7 @@ func (p *planner) maybeLogStatementInternal(
 	ctx context.Context,
 	execType executorType,
 	isCopy bool,
-	numRetries, txnCounter, rows int,
+	numRetries, txnCounter, rows, stmtCount int,
 	bulkJobId uint64,
 	err error,
 	startTime time.Time,
@@ -215,6 +215,7 @@ func (p *planner) maybeLogStatementInternal(
 		FullTableScan: p.curPlan.flags.IsSet(planFlagContainsFullTableScan),
 		FullIndexScan: p.curPlan.flags.IsSet(planFlagContainsFullIndexScan),
 		TxnCounter:    uint32(txnCounter),
+		StmtPosInTxn:  uint32(stmtCount),
 	}
 
 	// Note that for bulk job query (IMPORT, BACKUP and RESTORE), we don't