sql: step read timestamp before RC statement retries

In `sql.(*connExecutor).dispatchReadCommittedStmtToExecutionEngine` we
automatically retry statements under Read Committed isolation that fail
with retryable errors. To give these retries a better chance of
succeeding, step the read timestamp just before each retry. This should
help the retried query see the results of the conflicting transactions
that caused the failure and any other transactions that occurred in the
meantime.

Also step statement_timestamp so that any SQL using statement_timestamp() gets
an up-to-date timestamp on each retry.

Informs: #145377

Release note: None

Author: michae2

pkg/sql/conn_executor_exec.go

@@ -2560,18 +2560,18 @@ func (ex *connExecutor) commitSQLTransactionInternal(ctx context.Context) (retEr
 
 	ex.extraTxnState.prepStmtsNamespace.closePortals(ctx, &ex.extraTxnState.prepStmtsNamespaceMemAcc)
 
-	// We need to step the transaction's internal read sequence before committing
-	// if it has stepping enabled. If it doesn't have stepping enabled, then we
-	// just set the stepping mode back to what it was.
+	// We need to step the transaction's read sequence before committing if it has
+	// stepping enabled. If it doesn't have stepping enabled, then we just set the
+	// stepping mode back to what it was.
 	//
-	// Even if we do step the transaction's internal read sequence, we do not
-	// advance its external read timestamp (applicable only to read committed
-	// transactions). This is because doing so is not needed before committing,
-	// and it would cause the transaction to commit at a higher timestamp than
-	// necessary. On heavily contended workloads like the one from #109628, this
-	// can cause unnecessary write-write contention between transactions by
-	// inflating the contention footprint of each transaction (i.e. the duration
-	// measured in MVCC time that the transaction holds locks).
+	// Even if we do step the transaction's read sequence, we do not advance its
+	// read timestamp (applicable only to read committed transactions). This is
+	// because doing so is not needed before committing, and it would cause the
+	// transaction to commit at a higher timestamp than necessary. On heavily
+	// contended workloads like the one from #109628, this can cause unnecessary
+	// write-write contention between transactions by inflating the contention
+	// footprint of each transaction (i.e. the duration measured in MVCC time that
+	// the transaction holds locks).
 	prevSteppingMode := ex.state.mu.txn.ConfigureStepping(ctx, kv.SteppingEnabled)
 	if prevSteppingMode == kv.SteppingEnabled {
 		if err := ex.state.mu.txn.Step(ctx, false /* allowReadTimestampStep */); err != nil {
@@ -2722,6 +2722,14 @@ func (ex *connExecutor) rollbackSQLTransaction(
 func (ex *connExecutor) dispatchReadCommittedStmtToExecutionEngine(
 	ctx context.Context, p *planner, res RestrictedCommandResult,
 ) error {
+	if ex.executorType == executorTypeInternal {
+		// Because we step the read timestamp below, this is not safe to call within
+		// internal executor.
+		return errors.AssertionFailedf(
+			"call of dispatchReadCommittedStmtToExecutionEngine within internal executor",
+		)
+	}
+
 	getPausablePortalInfo := func() *portalPauseInfo {
 		if p != nil && p.pausablePortal != nil {
 			return p.pausablePortal.pauseInfo
@@ -2749,7 +2757,17 @@ func (ex *connExecutor) dispatchReadCommittedStmtToExecutionEngine(
 			ex.sessionTracing.TraceRetryInformation(
 				ctx, "statement", p.autoRetryStmtCounter, p.autoRetryStmtReason,
 			)
+			// Step both the sequence number and the read timestamp so that we can see
+			// the results of the conflicting transactions that caused us to fail and
+			// any other transactions that occurred in the meantime.
+			if err := ex.state.mu.txn.Step(ctx, true /* allowReadTimestampStep */); err != nil {
+				return err
+			}
+			// Also step statement_timestamp so that any SQL using it is up-to-date.
+			stmtTS := ex.server.cfg.Clock.PhysicalTime()
+			p.extendedEvalCtx.StmtTimestamp = stmtTS
 		}
+
 		bufferPos := res.BufferedResultsLen()
 		if err = ex.dispatchToExecutionEngine(ctx, p, res); err != nil {
 			return err
@@ -2802,9 +2820,6 @@ func (ex *connExecutor) dispatchReadCommittedStmtToExecutionEngine(
 		if err := ex.state.mu.txn.PrepareForPartialRetry(ctx); err != nil {
 			return err
 		}
-		if err := ex.state.mu.txn.Step(ctx, false /* allowReadTimestampStep */); err != nil {
-			return err
-		}
 		p.autoRetryStmtCounter++
 		p.autoRetryStmtReason = maybeRetriableErr
 		if ppInfo := getPausablePortalInfo(); ppInfo != nil {