kv: latching changes to support shared locks

Reads that acquire shared locks need to be isolated with concurrent
write requests. They also need to be isolated with exclusive-locking
reads. They can, however, run concurrently with other shared locking
reads and non-locking reads.

As described in the shared locks RFC, to make these semantics work,
shared-locking reads should acquire read latches at `hlc.MaxTimestamp`.
This patch makes that change and adds test cases for all these
interactions.

Closes #102264

Release note: None

pkg/kv/kvserver/batcheval/declare.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/errors"
 )
 
 // DefaultDeclareKeys is the default implementation of Command.DeclareKeys.
@@ -57,35 +58,65 @@ func DefaultDeclareIsolatedKeys(
 	maxOffset time.Duration,
 ) {
 	access := spanset.SpanReadWrite
-	// TODO(arul): pass in the correct lock strength here based on the request.
 	str := lock.Intent
 	timestamp := header.Timestamp
-	if kvpb.IsReadOnly(req) && !kvpb.IsLocking(req) {
-		access = spanset.SpanReadOnly
-		str = lock.None
-
-		// For non-locking reads, acquire read latches all the way up to the
-		// request's worst-case (i.e. global) uncertainty limit, because reads may
-		// observe writes all the way up to this timestamp.
-		//
-		// It is critical that reads declare latches up through their uncertainty
-		// interval so that they are properly synchronized with earlier writes that
-		// may have a happened-before relationship with the read. These writes could
-		// not have completed and returned to the client until they were durable in
-		// the Range's Raft log. However, they may not have been applied to the
-		// replica's state machine by the time the write was acknowledged, because
-		// Raft entry application occurs asynchronously with respect to the writer
-		// (see AckCommittedEntriesBeforeApplication). Latching is the only
-		// mechanism that ensures that any observers of the write wait for the write
-		// apply before reading.
-		//
-		// NOTE: we pass an empty lease status here, which means that observed
-		// timestamps collected by transactions will not be used. The actual
-		// uncertainty interval used by the request may be smaller (i.e. contain a
-		// local limit), but we can't determine that until after we have declared
-		// keys, acquired latches, and consulted the replica's lease.
-		in := uncertainty.ComputeInterval(header, kvserverpb.LeaseStatus{}, maxOffset)
-		timestamp.Forward(in.GlobalLimit)
+	if kvpb.IsReadOnly(req) {
+		if !kvpb.IsLocking(req) {
+			access = spanset.SpanReadOnly
+			str = lock.None
+			// For non-locking reads, acquire read latches all the way up to the
+			// request's worst-case (i.e. global) uncertainty limit, because reads may
+			// observe writes all the way up to this timestamp.
+			//
+			// It is critical that reads declare latches up through their uncertainty
+			// interval so that they are properly synchronized with earlier writes that
+			// may have a happened-before relationship with the read. These writes could
+			// not have completed and returned to the client until they were durable in
+			// the Range's Raft log. However, they may not have been applied to the
+			// replica's state machine by the time the write was acknowledged, because
+			// Raft entry application occurs asynchronously with respect to the writer
+			// (see AckCommittedEntriesBeforeApplication). Latching is the only
+			// mechanism that ensures that any observers of the write wait for the write
+			// apply before reading.
+			//
+			// NOTE: we pass an empty lease status here, which means that observed
+			// timestamps collected by transactions will not be used. The actual
+			// uncertainty interval used by the request may be smaller (i.e. contain a
+			// local limit), but we can't determine that until after we have declared
+			// keys, acquired latches, and consulted the replica's lease.
+			in := uncertainty.ComputeInterval(header, kvserverpb.LeaseStatus{}, maxOffset)
+			timestamp.Forward(in.GlobalLimit)
+		} else {
+			str = req.(kvpb.LockingReadRequest).KeyLockingStrength()
+			switch str {
+			// The lock.None case has already been handled above.
+			case lock.Shared:
+				access = spanset.SpanReadOnly
+				// Unlike non-locking reads, shared-locking reads are isolated from
+				// writes at all timestamps[1] (not just the request's timestamp); so we
+				// acquire a read latch at max timestamp. See the shared locks RFC for
+				// more details.
+				//
+				// [1] This allows the transaction that issued the shared-locking read to
+				// get arbitrarily bumped without needing to worry about its read-refresh
+				// failing. Unlike non-locking reads, where the read set is only protected
+				// from concurrent writers operating at lower timestamps, a shared-locking
+				// read extends this protection to all timestamps.
+				timestamp = hlc.MaxTimestamp
+			case lock.Exclusive:
+				// Reads that acquire exclusive locks acquire write latches at the
+				// request's timestamp. This isolates them from all concurrent writes,
+				// all concurrent shared-locking reads, and non-locking reads at higher
+				// timestamps[1].
+				//
+				// [1] This won't be required once non-locking reads stop blocking on
+				// exclusive locks. Once that happens, exclusive locks should acquire
+				// write latches at hlc.MaxTimestamp.
+				access = spanset.SpanReadWrite
+			default:
+				panic(errors.AssertionFailedf("unexpected lock strength %s", str))
+			}
+		}
 	}
 	latchSpans.AddMVCC(access, req.Header().Span(), timestamp)
 	lockSpans.Add(str, req.Header().Span())

pkg/kv/kvserver/concurrency/datadriven_util_test.go

@@ -15,6 +15,7 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/poison"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -169,12 +170,20 @@ func scanSingleRequest(
 		}
 		return enginepb.TxnSeq(n)
 	}
+	maybeGetStr := func() lock.Strength {
+		s, ok := fields["str"]
+		if !ok {
+			return lock.None
+		}
+		return concurrency.GetStrength(t, d, s)
+	}
 
 	switch cmd {
 	case "get":
 		var r kvpb.GetRequest
 		r.Sequence = maybeGetSeq()
 		r.Key = roachpb.Key(mustGetField("key"))
+		r.KeyLocking = maybeGetStr()
 		return &r
 
 	case "scan":
@@ -184,6 +193,7 @@ func scanSingleRequest(
 		if v, ok := fields["endkey"]; ok {
 			r.EndKey = roachpb.Key(v)
 		}
+		r.KeyLocking = maybeGetStr()
 		return &r
 
 	case "put":

pkg/kv/kvserver/concurrency/lock_table_test.go

@@ -707,7 +707,7 @@ func scanSpans(
 		}
 		strS := parts[0]
 		spanStr := parts[1]
-		str := getStrength(t, d, strS)
+		str := GetStrength(t, d, strS)
 		// Compute latch span access based on the supplied strength.
 		var sa spanset.SpanAccess
 		switch str {
@@ -756,10 +756,10 @@ func ScanIsoLevel(t *testing.T, d *datadriven.TestData) isolation.Level {
 func ScanLockStrength(t *testing.T, d *datadriven.TestData) lock.Strength {
 	var strS string
 	d.ScanArgs(t, "strength", &strS)
-	return getStrength(t, d, strS)
+	return GetStrength(t, d, strS)
 }
 
-func getStrength(t *testing.T, d *datadriven.TestData, strS string) lock.Strength {
+func GetStrength(t *testing.T, d *datadriven.TestData, strS string) lock.Strength {
 	switch strS {
 	case "none":
 		return lock.None