storage: More CheckSSTConflicts fixes

A few additional fixes around CheckSSTConflicts, stats
calculations, and Next()ing logic, caught by kvnemesis.
Hopefully the last of its kind.

Also re-enable kvnemesis testing for range keys in
AddSSTable, reverting #98475.

Fixes #94141.
Fixes #98473.
Informs #94876.

Epic: none

Release note: None

pkg/kv/kvnemesis/generator.go

@@ -525,12 +525,7 @@ func randAddSSTable(g *generator, rng *rand.Rand) Operation {
 	probTombstone := 0.2             // probability to write a tombstone
 	asWrites := rng.Float64() < 0.2  // IngestAsWrites
 
-	if true {
-		// TODO(erikgrinaker): Disable range keys for now since CheckSSTConflicts
-		// computes incorrect MVCC stats. See:
-		// https://github.com/cockroachdb/cockroach/issues/98473
-		numRangeKeys = 0
-	} else if r := rng.Float64(); r < 0.8 {
+	if r := rng.Float64(); r < 0.8 {
 		// 80% probability of only point keys.
 		numRangeKeys = 0
 	} else if r < 0.9 {

pkg/kv/kvserver/batcheval/cmd_add_sstable_test.go

@@ -851,6 +851,20 @@ func TestEvalAddSSTable(t *testing.T) {
 			sst:        kvs{rangeKV("a", "l", 8, "")},
 			expect:     kvs{rangeKV("a", "c", 8, ""), rangeKV("c", "d", 8, ""), rangeKV("c", "d", 6, ""), rangeKV("d", "j", 8, ""), rangeKV("j", "k", 8, ""), rangeKV("j", "k", 5, ""), rangeKV("k", "l", 8, "")},
 		},
+		"DisallowConflicts correctly accounts for complex fragment cases 5": {
+			noConflict: true,
+			reqTS:      10,
+			data:       kvs{pointKV("cc", 7, ""), pointKV("cc", 6, ""), pointKV("cc", 5, "foo"), pointKV("cc", 4, ""), pointKV("cc", 3, "bar"), pointKV("cc", 2, "barfoo"), rangeKV("ab", "g", 1, "")},
+			sst:        kvs{pointKV("aa", 8, "foo"), pointKV("aaa", 8, ""), pointKV("ac", 8, "foo"), rangeKV("b", "c", 8, ""), pointKV("ca", 8, "foo"), pointKV("cb", 8, "foo"), pointKV("cc", 8, "foo"), rangeKV("d", "e", 8, ""), pointKV("e", 8, "foobar")},
+			expect:     kvs{pointKV("aa", 8, "foo"), pointKV("aaa", 8, ""), rangeKV("ab", "b", 1, ""), pointKV("ac", 8, "foo"), rangeKV("b", "c", 8, ""), rangeKV("b", "c", 1, ""), rangeKV("c", "d", 1, ""), pointKV("ca", 8, "foo"), pointKV("cb", 8, "foo"), pointKV("cc", 8, "foo"), pointKV("cc", 7, ""), pointKV("cc", 6, ""), pointKV("cc", 5, "foo"), pointKV("cc", 4, ""), pointKV("cc", 3, "bar"), pointKV("cc", 2, "barfoo"), rangeKV("d", "e", 8, ""), rangeKV("d", "e", 1, ""), rangeKV("e", "g", 1, ""), pointKV("e", 8, "foobar")},
+		},
+		"DisallowConflicts handles existing point key above existing range tombstone": {
+			noConflict: true,
+			reqTS:      10,
+			data:       kvs{pointKV("c", 7, ""), rangeKV("a", "g", 6, ""), pointKV("h", 7, "")},
+			sst:        kvs{rangeKV("b", "d", 8, ""), rangeKV("f", "j", 8, "")},
+			expect:     kvs{rangeKV("a", "b", 6, ""), rangeKV("b", "d", 8, ""), rangeKV("b", "d", 6, ""), pointKV("c", 7, ""), rangeKV("d", "f", 6, ""), rangeKV("f", "g", 8, ""), rangeKV("f", "g", 6, ""), rangeKV("g", "j", 8, ""), pointKV("h", 7, "")},
+		},
 		"DisallowConflicts accounts for point key already deleted in engine": {
 			noConflict: true,
 			reqTS:      10,
@@ -1024,7 +1038,7 @@ func TestEvalAddSSTable(t *testing.T) {
 			noConflict: true,
 			data:       kvs{rangeKV("a", "b", 7, "")},
 			sst:        kvs{rangeKV("a", "b", 7, "")},
-			expectErr:  "ingested range key collides with an existing one",
+			expectErr:  &kvpb.WriteTooOldError{},
 		},
 	}
 	testutils.RunTrueAndFalse(t, "IngestAsWrites", func(t *testing.T, ingestAsWrites bool) {

pkg/storage/sst.go

@@ -473,8 +473,8 @@ func CheckSSTConflicts(
 					isIdempotent := extTombstones.Equal(sstRangeKeys.Versions)
 					if ok := allowIdempotentHelper(extRangeKeys.Versions[0].Timestamp); !ok || !isIdempotent {
 						// Idempotence is either not allowed or there's a conflict.
-						return enginepb.MVCCStats{}, errors.Errorf(
-							"ingested range key collides with an existing one: %s", sstTombstone)
+						return enginepb.MVCCStats{}, kvpb.NewWriteTooOldError(
+							sstTombstone.Timestamp, extRangeKeys.Versions[0].Timestamp.Next(), sstRangeKeys.Bounds.Key)
 					}
 				}
 			}
@@ -813,6 +813,21 @@ func CheckSSTConflicts(
 			// We exclude !sstHasPoint above in case we were at a range key pause
 			// point that matches extKey. In that case, the below SeekGE would make
 			// no forward progress.
+			//
+			// However, seeking is not safe if we're at an ext range key; we could
+			// miss conflicts and overlaps with sst range keys in between
+			// [current SST Key, extKey.Key). Do a next and go back to the start of
+			// the loop. If we had a dedicated sst range key iterator, we could have
+			// optimized away this unconditional-next.
+			if extHasRange {
+				sstIter.NextKey()
+				sstOK, sstErr = sstIter.Valid()
+				if sstOK {
+					extIter.SeekGE(MVCCKey{Key: sstIter.UnsafeKey().Key})
+					extOK, extErr = extIter.Valid()
+				}
+				continue
+			}
 			sstIter.SeekGE(MVCCKey{Key: extKey.Key})
 			sstOK, sstErr = sstIter.Valid()
 			if sstOK {
@@ -1040,10 +1055,10 @@ func CheckSSTConflicts(
 					// ext key.
 					extIter.NextKey()
 					extOK, extErr = extIter.Valid()
-					if extOK {
+					if extOK && extIter.UnsafeKey().Key.Compare(sstIter.UnsafeKey().Key) < 0 {
 						sstIter.SeekGE(MVCCKey{Key: extIter.UnsafeKey().Key})
+						sstOK, sstErr = sstIter.Valid()
 					}
-					sstOK, sstErr = sstIter.Valid()
 					if sstOK {
 						extIter.SeekGE(MVCCKey{Key: sstIter.UnsafeKey().Key})
 						extOK, extErr = extIter.Valid()