opt: add enable_durable_locking_for_serializable session variable

Follow-up from #105857

This commit ammends 6a3e43d to add a
session variable to control whether guaranteed-durable locks are used
under serializable isolation.

Informs: #100194

Epic: CRDB-25322

Release note (sql change): Add a new session variable,
`enable_durable_locking_for_serializable`, which controls locking
durability under serializable isolation. With this set to true, SELECT
FOR UPDATE locks, SELECT FOR SHARED locks, and constraint check locks
(e.g. locks acquired during foreign key checks if
`enable_implicit_fk_locking_for_serializable` is set to true) will be
guaranteed-durable under serializable isolation, meaning they will
always be held to transaction commit. (These locks are always
guaranteed-durable under weaker isolation levels.)

By default, under serializable isolation these locks are best-effort
rather than guaranteed-durable, meaning in some cases (e.g. leaseholder
transfer, node loss, etc.) they could be released before transaction
commit. Serializable isolation does not rely on locking for correctness,
only using it to improve performance under contention, so this default
is a deliberate choice to avoid the performance overhead of lock
replication.

pkg/sql/exec_util.go

@@ -3616,6 +3616,10 @@ func (m *sessionDataMutator) SetImplicitFKLockingForSerializable(val bool) {
 	m.data.ImplicitFKLockingForSerializable = val
 }
 
+func (m *sessionDataMutator) SetDurableLockingForSerializable(val bool) {
+	m.data.DurableLockingForSerializable = val
+}
+
 // Utility functions related to scrubbing sensitive information on SQL Stats.
 
 // quantizeCounts ensures that the Count field in the

pkg/sql/opt/exec/execbuilder/testdata/fk

@@ -77,6 +77,29 @@ vectorized: true
                   estimated row count: 2
                   label: buffer 1
 
+# Try again with durable locking enabled.
+statement ok
+SET enable_durable_locking_for_serializable = true
+
+# TODO(michae2, 100194): Change this from EXPLAIN (OPT) to EXPLAIN.
+query T
+EXPLAIN (OPT) INSERT INTO child VALUES (1,1), (2,2)
+----
+insert child
+ ├── values
+ │    ├── (1, 1)
+ │    └── (2, 2)
+ └── f-k-checks
+      └── f-k-checks-item: child(p) -> parent(p)
+           └── anti-join (lookup parent)
+                ├── lookup columns are key
+                ├── locking: for-share,durability-guaranteed
+                ├── with-scan &1
+                └── filters (true)
+
+statement ok
+RESET enable_durable_locking_for_serializable
+
 statement ok
 RESET enable_implicit_fk_locking_for_serializable
 
@@ -644,6 +667,30 @@ vectorized: true
                   spans: FULL SCAN
                   locking strength: for share
 
+# Try again with durable locking enabled.
+statement ok
+SET enable_durable_locking_for_serializable = true
+
+# TODO(michae2, 100194): Change this from EXPLAIN (OPT) to EXPLAIN.
+query T
+EXPLAIN (OPT) UPDATE child SET p = 4
+----
+update child
+ ├── project
+ │    ├── scan child
+ │    └── projections
+ │         └── 4
+ └── f-k-checks
+      └── f-k-checks-item: child(p) -> parent(p)
+           └── anti-join (merge)
+                ├── with-scan &1
+                ├── scan parent
+                │    └── locking: for-share,durability-guaranteed
+                └── filters (true)
+
+statement ok
+RESET enable_durable_locking_for_serializable
+
 statement ok
 RESET enable_implicit_fk_locking_for_serializable
 

pkg/sql/opt/exec/execbuilder/testdata/select_for_update

@@ -2560,3 +2560,84 @@ vectorized: true
           spans: /2-/3
           locking strength: for update
           locking wait policy: skip locked
+
+# ------------------------------------------------------------------------------
+# Tests with durable locking.
+# ------------------------------------------------------------------------------
+
+statement ok
+SET enable_durable_locking_for_serializable = true
+
+# TODO(michae2, 100194): Change these from EXPLAIN (OPT) to EXPLAIN (VERBOSE).
+
+query T
+EXPLAIN (OPT) SELECT * FROM t FOR UPDATE
+----
+scan t
+ └── locking: for-update,durability-guaranteed
+
+query T
+EXPLAIN (OPT) SELECT * FROM t FOR NO KEY UPDATE
+----
+scan t
+ └── locking: for-no-key-update,durability-guaranteed
+
+query T
+EXPLAIN (OPT) SELECT * FROM t FOR SHARE
+----
+scan t
+ └── locking: for-share,durability-guaranteed
+
+query T
+EXPLAIN (OPT) SELECT * FROM t FOR KEY SHARE
+----
+scan t
+ └── locking: for-key-share,durability-guaranteed
+
+query T
+EXPLAIN (OPT) SELECT * FROM t FOR KEY SHARE FOR SHARE
+----
+scan t
+ └── locking: for-share,durability-guaranteed
+
+query T
+EXPLAIN (OPT) SELECT * FROM t FOR KEY SHARE FOR SHARE FOR NO KEY UPDATE
+----
+scan t
+ └── locking: for-no-key-update,durability-guaranteed
+
+query T
+EXPLAIN (OPT) SELECT * FROM t FOR KEY SHARE FOR SHARE FOR NO KEY UPDATE FOR UPDATE
+----
+scan t
+ └── locking: for-update,durability-guaranteed
+
+query T
+EXPLAIN (OPT) SELECT * FROM t FOR UPDATE OF t
+----
+scan t
+ └── locking: for-update,durability-guaranteed
+
+query T
+EXPLAIN (OPT) SELECT (SELECT a FROM t FOR UPDATE OF t)
+----
+values
+ └── tuple
+      └── subquery
+           └── max1-row
+                └── scan t
+                     └── locking: for-update,durability-guaranteed
+
+query T
+EXPLAIN (OPT) SELECT * FROM t WHERE a IN (SELECT b FROM t FOR UPDATE)
+----
+project
+ └── inner-join (lookup t)
+      ├── lookup columns are key
+      ├── distinct-on
+      │    └── scan t
+      │         └── locking: for-update,durability-guaranteed
+      └── filters (true)
+
+statement ok
+RESET enable_durable_locking_for_serializable

pkg/sql/opt/memo/memo.go

@@ -170,6 +170,7 @@ type Memo struct {
 	useImprovedComputedColumnFiltersDerivation bool
 	useImprovedJoinElimination                 bool
 	implicitFKLockingForSerializable           bool
+	durableLockingForSerializable              bool
 
 	// txnIsoLevel is the isolation level under which the plan was created. This
 	// affects the planning of some locking operations, so it must be included in
@@ -238,6 +239,7 @@ func (m *Memo) Init(ctx context.Context, evalCtx *eval.Context) {
 		useImprovedComputedColumnFiltersDerivation: evalCtx.SessionData().OptimizerUseImprovedComputedColumnFiltersDerivation,
 		useImprovedJoinElimination:                 evalCtx.SessionData().OptimizerUseImprovedJoinElimination,
 		implicitFKLockingForSerializable:           evalCtx.SessionData().ImplicitFKLockingForSerializable,
+		durableLockingForSerializable:              evalCtx.SessionData().DurableLockingForSerializable,
 		txnIsoLevel:                                evalCtx.TxnIsoLevel,
 	}
 	m.metadata.Init()
@@ -386,6 +388,7 @@ func (m *Memo) IsStale(
 		m.useImprovedComputedColumnFiltersDerivation != evalCtx.SessionData().OptimizerUseImprovedComputedColumnFiltersDerivation ||
 		m.useImprovedJoinElimination != evalCtx.SessionData().OptimizerUseImprovedJoinElimination ||
 		m.implicitFKLockingForSerializable != evalCtx.SessionData().ImplicitFKLockingForSerializable ||
+		m.durableLockingForSerializable != evalCtx.SessionData().DurableLockingForSerializable ||
 		m.txnIsoLevel != evalCtx.TxnIsoLevel {
 		return true, nil
 	}

pkg/sql/opt/memo/memo_test.go

@@ -386,6 +386,12 @@ func TestMemoIsStale(t *testing.T) {
 	evalCtx.SessionData().ImplicitFKLockingForSerializable = false
 	notStale()
 
+	// Stale enable_durable_locking_for_serializable.
+	evalCtx.SessionData().DurableLockingForSerializable = true
+	stale()
+	evalCtx.SessionData().DurableLockingForSerializable = false
+	notStale()
+
 	// Stale txn isolation level.
 	evalCtx.TxnIsoLevel = isolation.ReadCommitted
 	stale()

pkg/sql/opt/optbuilder/select.go

@@ -697,16 +697,22 @@ func (b *Builder) buildScan(
 	}
 	if locking.isSet() {
 		private.Locking = locking.get()
-		if b.evalCtx.TxnIsoLevel != isolation.Serializable {
+		if b.evalCtx.TxnIsoLevel != isolation.Serializable ||
+			b.evalCtx.SessionData().DurableLockingForSerializable {
 			// Under weaker isolation levels we use fully-durable locks for SELECT FOR
-			// UPDATE statements, SELECT FOR SHARE statements, and all other locked
-			// scans (e.g. FK checks), regardless of locking strength and wait
-			// policy. Unlike mutation statements, SELECT FOR UPDATE statements do not
-			// lay down intents, so we cannot rely on the durability of intents to
+			// UPDATE statements, SELECT FOR SHARE statements, and constraint checks
+			// (e.g. FK checks), regardless of locking strength and wait policy.
+			// Unlike mutation statements, SELECT FOR UPDATE statements do not lay
+			// down intents, so we cannot rely on the durability of intents to
 			// guarantee exclusion until commit as we do for mutation statements. And
 			// unlike serializable isolation, weaker isolation levels do not perform
 			// read refreshing, so we cannot rely on read refreshing to guarantee
 			// exclusion.
+			//
+			// Under serializable isolation we only use fully-durable locks if
+			// enable_durable_locking_for_serializable is set. (Serializable isolation
+			// does not require locking for correctness, so by default we use
+			// best-effort locks for better performance.)
 			private.Locking.Durability = tree.LockDurabilityGuaranteed
 		}
 		if private.Locking.WaitPolicy == tree.LockWaitSkipLocked && tab.FamilyCount() > 1 {

pkg/sql/sessiondatapb/local_only_session_data.proto

@@ -417,6 +417,12 @@ message LocalOnlySessionData {
   bool implicit_fk_locking_for_serializable = 108 [
     (gogoproto.customname) = "ImplicitFKLockingForSerializable"
   ];
+  // DurableLockingForSerializable is true if we should use durable locking for
+  // SELECT FOR UPDATE statements, SELECT FOR SHARE statements, and constraint
+  // checks under serializable isolation. (Serializable isolation does not
+  // require locking for correctness, so by default we use best-effor locks for
+  // better performance.) Weaker isolation levels always use durable locking.
+  bool durable_locking_for_serializable = 109;
 
   ///////////////////////////////////////////////////////////////////////////
   // WARNING: consider whether a session parameter you're adding needs to  //

pkg/sql/vars.go

@@ -2834,6 +2834,23 @@ var varGen = map[string]sessionVar{
 		},
 		GlobalDefault: globalFalse,
 	},
+
+	// CockroachDB extension.
+	`enable_durable_locking_for_serializable`: {
+		GetStringVal: makePostgresBoolGetStringValFn(`enable_durable_locking_for_serializable`),
+		Set: func(_ context.Context, m sessionDataMutator, s string) error {
+			b, err := paramparse.ParseBoolVar("enable_durable_locking_for_serializable", s)
+			if err != nil {
+				return err
+			}
+			m.SetDurableLockingForSerializable(b)
+			return nil
+		},
+		Get: func(evalCtx *extendedEvalContext, _ *kv.Txn) (string, error) {
+			return formatBoolAsPostgresSetting(evalCtx.SessionData().DurableLockingForSerializable), nil
+		},
+		GlobalDefault: globalFalse,
+	},
 }
 
 func ReplicationModeFromString(s string) (sessiondatapb.ReplicationMode, error) {