sql: add exponential backoff to read committed stmt retry loop

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 READ COMMITTED 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`).

Author: michae2

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"
@@ -2746,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)
@@ -2767,7 +2779,6 @@ func (ex *connExecutor) dispatchReadCommittedStmtToExecutionEngine(
 			stmtTS := ex.server.cfg.Clock.PhysicalTime()
 			p.extendedEvalCtx.StmtTimestamp = stmtTS
 		}
-
 		bufferPos := res.BufferedResultsLen()
 		if err = ex.dispatchToExecutionEngine(ctx, p, res); err != nil {
 			return err
@@ -2828,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) {