kv: disable use of shared locks in conjunction with skip locked

We currently don't support the use of shared locks and the skip locked
wait policy together. Until this limitation is lifted, we should return
unimplemented errors instead of the wrong result.

Informs #110743

Release note: None

pkg/kv/kvnemesis/BUILD.bazel

@@ -27,6 +27,7 @@ go_library(
         "//pkg/kv/kvnemesis/kvnemesisutil",
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver",
+        "//pkg/kv/kvserver/batcheval",
         "//pkg/kv/kvserver/concurrency",
         "//pkg/kv/kvserver/concurrency/isolation",
         "//pkg/kv/kvserver/concurrency/lock",

pkg/kv/kvnemesis/applier.go

@@ -12,6 +12,7 @@ package kvnemesis
 
 import (
 	"context"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/batcheval"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/config/zonepb"
@@ -119,6 +120,10 @@ func exceptSkipLockedReplayError(err error) bool { // true if skip locked replay
 	return errors.Is(err, &concurrency.SkipLockedReplayError{})
 }
 
+func exceptSkipLockedUnsupportedError(err error) bool { // true if unsupported use of skip locked error
+	return errors.Is(err, &batcheval.SkipLockedUnsupportedError{})
+}
+
 func applyOp(ctx context.Context, env *Env, db *kv.DB, op *Operation) {
 	switch o := op.GetValue().(type) {
 	case *GetOperation,

pkg/kv/kvnemesis/validator.go

@@ -770,6 +770,7 @@ func (v *validator) processOp(op Operation) {
 		v.failIfError(
 			op, t.Result,
 			exceptRollback, exceptAmbiguous, exceptSharedLockPromotionError, exceptSkipLockedReplayError,
+			exceptSkipLockedUnsupportedError,
 		)
 
 		ops := t.Ops
@@ -1273,6 +1274,7 @@ func (v *validator) checkError(
 		exceptDelRangeUsingTombstoneStraddlesRangeBoundary,
 		exceptSharedLockPromotionError,
 		exceptSkipLockedReplayError,
+		exceptSkipLockedUnsupportedError,
 	}
 	sl = append(sl, extraExceptions...)
 	return v.failIfError(op, r, sl...)

pkg/kv/kvserver/batcheval/cmd_get.go

@@ -33,6 +33,10 @@ func Get(
 	h := cArgs.Header
 	reply := resp.(*kvpb.GetResponse)
 
+	if err := maybeDisallowSkipLockedRequest(h, args.KeyLockingStrength); err != nil {
+		return result.Result{}, err
+	}
+
 	getRes, err := storage.MVCCGet(ctx, readWriter, args.Key, h.Timestamp, storage.MVCCGetOptions{
 		Inconsistent:          h.ReadConsistency != kvpb.CONSISTENT,
 		SkipLocked:            h.WaitPolicy == lock.WaitPolicy_SkipLocked,

pkg/kv/kvserver/batcheval/cmd_reverse_scan.go

@@ -35,6 +35,10 @@ func ReverseScan(
 	h := cArgs.Header
 	reply := resp.(*kvpb.ReverseScanResponse)
 
+	if err := maybeDisallowSkipLockedRequest(h, args.KeyLockingStrength); err != nil {
+		return result.Result{}, err
+	}
+
 	var res result.Result
 	var scanRes storage.MVCCScanResult
 	var err error

pkg/kv/kvserver/batcheval/cmd_scan.go

@@ -13,12 +13,13 @@ package batcheval
 import (
 	"context"
 	"fmt"
-
+	"github.com/cockroachdb/cockroach/pkg/build"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/batcheval/result"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/util/admission/admissionpb"
+	"github.com/cockroachdb/errors"
 )
 
 func init() {
@@ -36,6 +37,10 @@ func Scan(
 	h := cArgs.Header
 	reply := resp.(*kvpb.ScanResponse)
 
+	if err := maybeDisallowSkipLockedRequest(h, args.KeyLockingStrength); err != nil {
+		return result.Result{}, err
+	}
+
 	var res result.Result
 	var scanRes storage.MVCCScanResult
 	var err error
@@ -133,3 +138,36 @@ func ScanReadCategory(ah kvpb.AdmissionHeader) storage.ReadCategory {
 	}
 	return readCategory
 }
+
+// maybeDisallowSkipLockedRequest returns an error if the skip locked wait
+// policy is used in conjunction with shared locks.
+//
+// TODO(arul): this won't be needed once
+// https://github.com/cockroachdb/cockroach/issues/110743 is addressed. Until
+// then, we return unimplemented errors.
+func maybeDisallowSkipLockedRequest(h kvpb.Header, str lock.Strength) error {
+	if h.WaitPolicy == lock.WaitPolicy_SkipLocked && str == lock.Shared {
+		return MarkSkipLockedUnsupportedError(errors.UnimplementedError(
+			errors.IssueLink{IssueURL: build.MakeIssueURL(110743)},
+			"shared locks are incompatible with the skip locked wait policy",
+		))
+	}
+	return nil
+}
+
+// SkipLockedUnsupportedError is used to mark errors resulting from unsupported
+// (currently unimplemented) uses of the skip locked wait policy.
+type SkipLockedUnsupportedError struct{}
+
+func (e *SkipLockedUnsupportedError) Error() string {
+	return "unsupported skip locked use error"
+}
+
+// MarkSkipLockedUnsupportedError wraps the given error, if not nil, as a skip
+// locked unsupported error.
+func MarkSkipLockedUnsupportedError(cause error) error {
+	if cause == nil {
+		return nil
+	}
+	return errors.Mark(cause, &SkipLockedUnsupportedError{})
+}

pkg/sql/logictest/testdata/logic_test/select_for_share

@@ -104,6 +104,12 @@ user testuser
 statement ok
 COMMIT
 
+statement ok
+SET enable_shared_locking_for_serializable = true
+
+statement error shared locks are incompatible with the skip locked wait policy
+SELECT * FROM t FOR SHARE SKIP LOCKED
+
 # TODO(arul): Add a test to show that the session setting doesn't apply to read
 # committed transactions. We currently can't issue SELECT FOR SHARE statements
 # in read committed transactions because durable locking hasn't been fully