kvnemesis: add CPut

This commit adds the following CPut operations to kvnemesis:

- `CPutMatchExisting`: the CPut condition matches an existing key's
  value. Both the write and the preceding read are validated.
- `CPutMatchMissing`: the CPut condition matches a missing key's nil
  value. This corresponds to the CPut generated by a typical `INSERT`
  statement. Again both the write and the preceding read are
  validated.
- `CPutNoMatch`: the CPut condition is not satisfied and the
  request/batch/transaction fails. These errors are expected and
  ignored by the test, but we still do some validation on the observed
  values for the condition to fail.
- `CPutAllowIfDoesNotExist`: the CPut condition is not satisfied but the
  operation succeeds if the value does not exist. The test is set up in
  a way that the value most likely does not exist.

Fixes: #64810

Release note: None

Author: miraradeva

pkg/kv/kvnemesis/BUILD.bazel

@@ -55,6 +55,7 @@ go_library(
         "@com_github_cockroachdb_pebble//:pebble",
         "@com_github_cockroachdb_pebble//vfs",
         "@org_golang_google_protobuf//proto",
+        "@org_golang_x_exp//maps",
     ],
 )
 
@@ -112,6 +113,7 @@ go_test(
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@org_golang_google_protobuf//proto",
+        "@org_golang_x_exp//maps",
     ],
 )
 

pkg/kv/kvnemesis/applier.go

@@ -105,10 +105,15 @@ func exceptDelRangeUsingTombstoneStraddlesRangeBoundary(err error) bool {
 	return errors.Is(err, errDelRangeUsingTombstoneStraddlesRangeBoundary)
 }
 
+func exceptConditionFailed(err error) bool {
+	return errors.HasType(err, (*kvpb.ConditionFailedError)(nil))
+}
+
 func applyOp(ctx context.Context, env *Env, db *kv.DB, op *Operation) {
 	switch o := op.GetValue().(type) {
 	case *GetOperation,
 		*PutOperation,
+		*CPutOperation,
 		*ScanOperation,
 		*BatchOperation,
 		*DeleteOperation,
@@ -342,6 +347,23 @@ func applyClientOp(
 			return
 		}
 		o.Result.OptionalTimestamp = ts
+	case *CPutOperation:
+		_, ts, err := dbRunWithResultAndTimestamp(ctx, db, func(b *kv.Batch) {
+			expVal := roachpb.MakeValueFromBytes(o.ExpVal)
+			if o.AllowIfDoesNotExist {
+				b.CPutAllowingIfNotExists(o.Key, o.Value(), expVal.TagAndDataBytes())
+			} else {
+				b.CPut(o.Key, o.Value(), expVal.TagAndDataBytes())
+			}
+			setLastReqSeq(b, o.Seq)
+		})
+		o.Result = resultInit(ctx, err)
+		// If the CPut failed with ConditionFailedError, we still want to record the
+		// timestamp and do some validation later.
+		if err != nil && !exceptConditionFailed(err) {
+			return
+		}
+		o.Result.OptionalTimestamp = ts
 	case *ScanOperation:
 		res, ts, err := dbRunWithResultAndTimestamp(ctx, db, func(b *kv.Batch) {
 			if o.SkipLocked {
@@ -561,6 +583,14 @@ func applyBatchOp(
 				b.Put(subO.Key, subO.Value())
 			}
 			setLastReqSeq(b, subO.Seq)
+		case *CPutOperation:
+			expVal := roachpb.MakeValueFromBytes(subO.ExpVal)
+			if subO.AllowIfDoesNotExist {
+				b.CPutAllowingIfNotExists(subO.Key, subO.Value(), expVal.TagAndDataBytes())
+			} else {
+				b.CPut(subO.Key, subO.Value(), expVal.TagAndDataBytes())
+			}
+			setLastReqSeq(b, subO.Seq)
 		case *ScanOperation:
 			if subO.SkipLocked {
 				panic(errors.AssertionFailedf(`SkipLocked cannot be used in batches`))
@@ -643,6 +673,10 @@ func applyBatchOp(
 			res := b.Results[resultIdx]
 			subO.Result = resultInit(ctx, res.Err)
 			subO.Result.ResumeSpan = res.ResumeSpan
+		case *CPutOperation:
+			res := b.Results[resultIdx]
+			subO.Result = resultInit(ctx, res.Err)
+			subO.Result.ResumeSpan = res.ResumeSpan
 		case *ScanOperation:
 			res := b.Results[resultIdx]
 			if res.Err != nil {