sql/opt: add locking durability

In addition to strength and wait policy, we now add a third property to
locks: durability. Locks with `LockDurabilityGuaranteed` are guaranteed
to be held until commit (if the transaction commits). Durable locks must
be used when correctness depends on locking. This is never the case
under our `SERIALIZABLE` isolation, but under `SNAPSHOT` and `READ
COMMITTED` isolation it will be the case for `SELECT FOR UPDATE`
statements, which will be the first users of durable locks.

This commit adds locking durability to the optimizer and `EXPLAIN`
output, but does not plumb it into KV yet. It will be used in the next
commit to temporarily disallow `SELECT FOR UPDATE` statements under
`SNAPSHOT` and `READ COMMITTED` isolation.

Release note: None

pkg/sql/catalog/descpb/locking.proto

@@ -115,7 +115,7 @@ enum ScanLockingStrength {
   // on each key scanned.
   FOR_UPDATE = 4;
 }
-  
+
 // LockingWaitPolicy controls the policy used for handling conflicting locks
 // held by other active transactions when attempting to lock rows due to FOR
 // UPDATE/SHARE clauses (i.e. it represents the NOWAIT and SKIP LOCKED options).

pkg/sql/opt/exec/execbuilder/mutation.go

@@ -973,8 +973,12 @@ func (b *Builder) canAutoCommit(rel memo.RelExpr) bool {
 
 // forUpdateLocking is the row-level locking mode used by mutations during their
 // initial row scan, when such locking is deemed desirable. The locking mode is
-// equivalent that used by a SELECT ... FOR UPDATE statement.
-var forUpdateLocking = opt.Locking{Strength: tree.ForUpdate}
+// equivalent to that used by a SELECT FOR UPDATE statement, except not durable.
+var forUpdateLocking = opt.Locking{
+	Strength:   tree.ForUpdate,
+	WaitPolicy: tree.LockWaitBlock,
+	Durability: tree.LockDurabilityBestEffort,
+}
 
 // shouldApplyImplicitLockingToMutationInput determines whether or not the
 // builder should apply a FOR UPDATE row-level locking mode to the initial row

pkg/sql/opt/exec/execbuilder/relational.go

@@ -620,7 +620,7 @@ func (b *Builder) scanParams(
 
 	locking := scan.Locking
 	if b.forceForUpdateLocking {
-		locking = forUpdateLocking
+		locking = locking.Max(forUpdateLocking)
 	}
 	b.ContainsNonDefaultKeyLocking = b.ContainsNonDefaultKeyLocking || locking.IsLocking()
 
@@ -2172,7 +2172,7 @@ func (b *Builder) buildIndexJoin(join *memo.IndexJoinExpr) (execPlan, error) {
 
 	locking := join.Locking
 	if b.forceForUpdateLocking {
-		locking = forUpdateLocking
+		locking = locking.Max(forUpdateLocking)
 	}
 	b.ContainsNonDefaultKeyLocking = b.ContainsNonDefaultKeyLocking || locking.IsLocking()
 
@@ -2493,7 +2493,7 @@ func (b *Builder) buildLookupJoin(join *memo.LookupJoinExpr) (execPlan, error) {
 
 	locking := join.Locking
 	if b.forceForUpdateLocking {
-		locking = forUpdateLocking
+		locking = locking.Max(forUpdateLocking)
 	}
 	b.ContainsNonDefaultKeyLocking = b.ContainsNonDefaultKeyLocking || locking.IsLocking()
 
@@ -2735,7 +2735,7 @@ func (b *Builder) buildInvertedJoin(join *memo.InvertedJoinExpr) (execPlan, erro
 
 	locking := join.Locking
 	if b.forceForUpdateLocking {
-		locking = forUpdateLocking
+		locking = locking.Max(forUpdateLocking)
 	}
 	b.ContainsNonDefaultKeyLocking = b.ContainsNonDefaultKeyLocking || locking.IsLocking()
 
@@ -2816,8 +2816,8 @@ func (b *Builder) buildZigzagJoin(join *memo.ZigzagJoinExpr) (execPlan, error) {
 	leftLocking := join.LeftLocking
 	rightLocking := join.RightLocking
 	if b.forceForUpdateLocking {
-		leftLocking = forUpdateLocking
-		rightLocking = forUpdateLocking
+		leftLocking = leftLocking.Max(forUpdateLocking)
+		rightLocking = rightLocking.Max(forUpdateLocking)
 	}
 	b.ContainsNonDefaultKeyLocking = b.ContainsNonDefaultKeyLocking || leftLocking.IsLocking() || rightLocking.IsLocking()