Merge #146860

146860: sql: add exponential backoff to read committed stmt retry loop r=yuzefovich,rafiss a=michae2

Testing has shown that adding exponential backoff significantly improves throughput of highly contentious read committed workloads.

Informs: #145377

Release note (sql change): Add session variable `initial_retry_backoff_for_read_committed` which controls the initial backoff duration when retrying an individual statement in an explicit READ COMMITTED transaction. A duration of 0 disables exponential backoff.

If a statement in an explicit transaction is failing with the following 40001 error:

```
ERROR: restart transaction: read committed retry limit exceeded; set by max_retries_for_read_committed=...
```

Then `initial_retry_backoff_for_read_committed` should be set to a duration proportional to the typical execution time of the statement (in addition to also increasing `max_retries_for_read_committed`).

Co-authored-by: Michael Erickson <[email protected]>

Author: craig[bot]

pkg/sql/conn_executor_exec.go

@@ -69,6 +69,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log/severity"
 	"github.com/cockroachdb/cockroach/pkg/util/metamorphic"
 	"github.com/cockroachdb/cockroach/pkg/util/metric"
+	"github.com/cockroachdb/cockroach/pkg/util/retry"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing/tracingpb"
@@ -2560,18 +2561,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 +2723,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
@@ -2738,8 +2747,19 @@ func (ex *connExecutor) dispatchReadCommittedStmtToExecutionEngine(
 		return err
 	}
 
+	// Use retry with exponential backoff and full jitter to reduce collisions for
+	// high-contention workloads. See https://en.wikipedia.org/wiki/Exponential_backoff and
+	// https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/
 	maxRetries := int(ex.sessionData().MaxRetriesForReadCommitted)
-	for attemptNum := 0; ; attemptNum++ {
+	initialBackoff := ex.sessionData().InitialRetryBackoffForReadCommitted
+	useBackoff := initialBackoff > 0
+	opts := retry.Options{
+		InitialBackoff:      initialBackoff,
+		MaxBackoff:          1024 * initialBackoff,
+		Multiplier:          2.0,
+		RandomizationFactor: 1.0,
+	}
+	for attemptNum, r := 0, retry.StartWithCtx(ctx, opts); !useBackoff || r.Next(); attemptNum++ {
 		// TODO(99410): Fix the phase time for pausable portals.
 		startExecTS := crtime.NowMono()
 		ex.statsCollector.PhaseTimes().SetSessionPhaseTime(sessionphase.PlannerMostRecentStartExecStmt, startExecTS)
@@ -2749,6 +2769,15 @@ 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 {
@@ -2802,9 +2831,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 {
@@ -2813,6 +2839,14 @@ func (ex *connExecutor) dispatchReadCommittedStmtToExecutionEngine(
 		}
 		ex.metrics.EngineMetrics.StatementRetryCount.Inc(1)
 	}
+	// Check if we exited the loop due to cancelation.
+	if useBackoff {
+		select {
+		case <-ctx.Done():
+			res.SetError(cancelchecker.QueryCanceledError)
+		default:
+		}
+	}
 	return nil
 }
 

pkg/sql/exec_util.go

@@ -4186,6 +4186,10 @@ func (m *sessionDataMutator) SetOptimizerUseExistsFilterHoistRule(val bool) {
 	m.data.OptimizerUseExistsFilterHoistRule = val
 }
 
+func (m *sessionDataMutator) SetInitialRetryBackoffForReadCommitted(val time.Duration) {
+	m.data.InitialRetryBackoffForReadCommitted = val
+}
+
 // Utility functions related to scrubbing sensitive information on SQL Stats.
 
 // quantizeCounts ensures that the Count field in the

pkg/sql/explain_bundle.go

@@ -1096,7 +1096,7 @@ func (c *stmtEnvCollector) PrintSessionSettings(w io.Writer, sv *settings.Values
 			switch varName {
 			case "idle_in_session_timeout", "idle_in_transaction_session_timeout",
 				"idle_session_timeout", "lock_timeout", "deadlock_timeout",
-				"statement_timeout", "transaction_timeout":
+				"statement_timeout", "transaction_timeout", "initial_retry_backoff_for_read_committed":
 				// Defaults for timeout settings are of the duration type (i.e.
 				// "0s"), so we'll parse it to extract the number of
 				// milliseconds (which is what the session variable uses).

pkg/sql/logictest/testdata/logic_test/information_schema

@@ -4011,6 +4011,7 @@ idle_in_transaction_session_timeout                        0
 idle_session_timeout                                       0
 index_join_streamer_batch_size                             8.0 MiB
 index_recommendations_enabled                              off
+initial_retry_backoff_for_read_committed                   0
 inject_retry_errors_enabled                                off
 inject_retry_errors_on_commit_enabled                      off
 integer_datetimes                                          on

pkg/sql/logictest/testdata/logic_test/pg_catalog

@@ -3005,6 +3005,7 @@ idle_in_transaction_session_timeout                        0                   N
 idle_session_timeout                                       0                   NULL      NULL        NULL        string
 index_join_streamer_batch_size                             8.0 MiB             NULL      NULL        NULL        string
 index_recommendations_enabled                              off                 NULL      NULL        NULL        string
+initial_retry_backoff_for_read_committed                   0                   NULL      NULL        NULL        string
 inject_retry_errors_enabled                                off                 NULL      NULL        NULL        string
 inject_retry_errors_on_commit_enabled                      off                 NULL      NULL        NULL        string
 integer_datetimes                                          on                  NULL      NULL        NULL        string
@@ -3235,6 +3236,7 @@ idle_in_transaction_session_timeout                        0                   N
 idle_session_timeout                                       0                   NULL  user     NULL      0s                  0s
 index_join_streamer_batch_size                             8.0 MiB             NULL  user     NULL      8.0 MiB             8.0 MiB
 index_recommendations_enabled                              off                 NULL  user     NULL      on                  false
+initial_retry_backoff_for_read_committed                   0                   NULL  user     NULL      0s                  0s
 inject_retry_errors_enabled                                off                 NULL  user     NULL      off                 off
 inject_retry_errors_on_commit_enabled                      off                 NULL  user     NULL      off                 off
 integer_datetimes                                          on                  NULL  user     NULL      on                  on
@@ -3453,6 +3455,7 @@ idle_in_transaction_session_timeout                        NULL    NULL     NULL
 idle_session_timeout                                       NULL    NULL     NULL     NULL        NULL
 index_join_streamer_batch_size                             NULL    NULL     NULL     NULL        NULL
 index_recommendations_enabled                              NULL    NULL     NULL     NULL        NULL
+initial_retry_backoff_for_read_committed                   NULL    NULL     NULL     NULL        NULL
 inject_retry_errors_enabled                                NULL    NULL     NULL     NULL        NULL
 inject_retry_errors_on_commit_enabled                      NULL    NULL     NULL     NULL        NULL
 integer_datetimes                                          NULL    NULL     NULL     NULL        NULL

pkg/sql/logictest/testdata/logic_test/show_source

@@ -119,6 +119,7 @@ idle_in_transaction_session_timeout                        0
 idle_session_timeout                                       0
 index_join_streamer_batch_size                             8.0 MiB
 index_recommendations_enabled                              off
+initial_retry_backoff_for_read_committed                   0
 inject_retry_errors_enabled                                off
 inject_retry_errors_on_commit_enabled                      off
 integer_datetimes                                          on

pkg/sql/sessiondatapb/local_only_session_data.proto

@@ -670,6 +670,10 @@ message LocalOnlySessionData {
   // OptimizerUseExistsFilterHoistRule, when true, causes the optimizer to apply
   // the HoistUnboundFilterFromExistsSubquery rule to EXISTS conditions.
   bool optimizer_use_exists_filter_hoist_rule = 170;
+  // InitialRetryBackoffForReadCommitted controls the initial backoff
+  // duration for automatic retries of statements in explicit READ COMMITTED
+  // transactions that see a transaction retry error. 0 disables backoff.
+  int64 initial_retry_backoff_for_read_committed = 171 [(gogoproto.casttype) = "time.Duration"];
   ///////////////////////////////////////////////////////////////////////////
   // WARNING: consider whether a session parameter you're adding needs to  //
   // be propagated to the remote nodes. If so, that parameter should live  //

pkg/sql/vars.go

@@ -2248,8 +2248,10 @@ var varGen = map[string]sessionVar{
 	},
 
 	// CockroachDB extension. Configures the maximum number of automatic retries
-	// to perform for statements in explicit READ COMMITTED transactions that
-	// see a transaction retry error.
+	// to perform for statements in explicit READ COMMITTED transactions that see
+	// a transaction retry error. (See also
+	// initial_retry_backoff_for_read_committed which should be tuned with
+	// max_retries_for_read_committed.)
 	`max_retries_for_read_committed`: {
 		GetStringVal: makeIntGetStringValFn(`max_retries_for_read_committed`),
 		Set: func(_ context.Context, m sessionDataMutator, s string) error {
@@ -4034,6 +4036,33 @@ var varGen = map[string]sessionVar{
 		},
 		GlobalDefault: globalTrue,
 	},
+
+	// CockroachDB extension. Configures the initial backoff duration for
+	// automatic retries of statements in explicit READ COMMITTED transactions
+	// that see a transaction retry error. For statements experiencing contention
+	// under READ COMMITTED isolation, this should be set to a duration
+	// proportional to the typical execution time of the statement (in addition to
+	// also increasing `max_retries_for_read_committed`).
+	`initial_retry_backoff_for_read_committed`: {
+		GetStringVal: makeTimeoutVarGetter(`initial_retry_backoff_for_read_committed`),
+		Set: func(_ context.Context, m sessionDataMutator, s string) error {
+			duration, err := validateTimeoutVar(m.data.GetIntervalStyle(), s,
+				"initial_retry_backoff_for_read_committed",
+			)
+			if err != nil {
+				return err
+			}
+			m.SetInitialRetryBackoffForReadCommitted(duration)
+			return nil
+		},
+		Get: func(evalCtx *extendedEvalContext, _ *kv.Txn) (string, error) {
+			ms := evalCtx.SessionData().InitialRetryBackoffForReadCommitted.Nanoseconds() / int64(time.Millisecond)
+			return strconv.FormatInt(ms, 10), nil
+		},
+		GlobalDefault: func(sv *settings.Values) string {
+			return "0s"
+		},
+	},
 }
 
 func ReplicationModeFromString(s string) (sessiondatapb.ReplicationMode, error) {

pkg/workload/kv/kv.go

@@ -93,6 +93,7 @@ type kv struct {
 	insertCount                          int
 	txnQoS                               string
 	prepareReadOnly                      bool
+	writesUseSelect1                     bool
 }
 
 func init() {
@@ -129,6 +130,7 @@ var kvMeta = workload.Meta{
 			`scatter`:           {RuntimeOnly: true},
 			`timeout`:           {RuntimeOnly: true},
 			`prepare-read-only`: {RuntimeOnly: true},
+			`sel1-writes`:       {RuntimeOnly: true},
 		}
 		g.flags.IntVar(&g.batchSize, `batch`, 1,
 			`Number of blocks to read/insert in a single SQL statement.`)
@@ -178,11 +180,13 @@ var kvMeta = workload.Meta{
 		g.flags.IntVar(&g.keySize, `key-size`, 0,
 			`Use string key of appropriate size instead of int`)
 		g.flags.DurationVar(&g.sfuDelay, `sfu-wait-delay`, 10*time.Millisecond,
-			`Delay before sfu write transaction commits or aborts`)
+			`Delay after SFU when using --sfu-writes (or after SELECT 1 when using --sel1-writes).`)
 		g.flags.StringVar(&g.txnQoS, `txn-qos`, `regular`,
 			`Set default_transaction_quality_of_service session variable, accepted`+
 				`values are 'background', 'regular' and 'critical'.`)
 		g.flags.BoolVar(&g.prepareReadOnly, `prepare-read-only`, false, `Prepare and perform only read statements.`)
+		g.flags.BoolVar(&g.writesUseSelect1, `sel1-writes`, false,
+			`Use SELECT 1 as the first statement of transactional writes with a sleep after SELECT 1.`)
 		g.connFlags = workload.NewConnFlags(&g.flags)
 		return g
 	},
@@ -550,6 +554,7 @@ func (w *kv) Ops(
 		if len(sfuStmtStr) > 0 && !op.config.prepareReadOnly {
 			op.sfuStmt = op.sr.Define(sfuStmtStr)
 		}
+		op.sel1Stmt = op.sr.Define("SELECT 1")
 		op.spanStmt = op.sr.Define(spanStmtStr)
 		if w.txnQoS != `regular` {
 			stmt := op.sr.Define(fmt.Sprintf(
@@ -582,6 +587,7 @@ type kvOp struct {
 	writeStmt        workload.StmtHandle
 	spanStmt         workload.StmtHandle
 	sfuStmt          workload.StmtHandle
+	sel1Stmt         workload.StmtHandle
 	delStmt          workload.StmtHandle
 	g                keyGenerator
 	t                keyTransformer
@@ -680,7 +686,7 @@ func (o *kvOp) run(ctx context.Context) (retErr error) {
 	}
 	start := timeutil.Now()
 	var err error
-	if o.config.writesUseSelectForUpdate {
+	if o.config.writesUseSelect1 || o.config.writesUseSelectForUpdate {
 		// We could use crdb.ExecuteTx, but we avoid retries in this workload so
 		// that each run call makes 1 attempt, so that rate limiting in workerRun
 		// behaves as expected.
@@ -696,15 +702,27 @@ func (o *kvOp) run(ctx context.Context) (retErr error) {
 				retErr = errors.CombineErrors(retErr, rollbackErr)
 			}
 		}()
-		rows, err := o.sfuStmt.QueryTx(ctx, tx, sfuArgs...)
-		if err != nil {
-			return err
+		if o.config.writesUseSelect1 {
+			rows, err := o.sel1Stmt.QueryTx(ctx, tx)
+			if err != nil {
+				return err
+			}
+			rows.Close()
+			if err = rows.Err(); err != nil {
+				return err
+			}
 		}
-		rows.Close()
-		if err = rows.Err(); err != nil {
-			return err
+		if o.config.writesUseSelectForUpdate {
+			rows, err := o.sfuStmt.QueryTx(ctx, tx, sfuArgs...)
+			if err != nil {
+				return err
+			}
+			rows.Close()
+			if err = rows.Err(); err != nil {
+				return err
+			}
 		}
-		// Simulate a transaction that does other work between the SFU and write.
+		// Simulate a transaction that does other work between the sel1 / SFU and write.
 		time.Sleep(o.config.sfuDelay)
 		if _, err = o.writeStmt.ExecTx(ctx, tx, writeArgs...); err != nil {
 			// Multiple write transactions can contend and encounter