kv: use version-less key in rditer.KeyRange

This commit replaces the use of `storage.MVCCKey` bounds with `roachpb.Key`
bounds in `rditer.KeyRange`. It addresses an existing TODO and helps clarify
things in #72121 slightly, as that PR is going to add a new field to `MVCCKey`.

pkg/kv/kvserver/client_merge_test.go

@@ -3789,7 +3789,7 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 		for _, r := range keyRanges {
 			sstFile := &storage.MemFile{}
 			sst := storage.MakeIngestionSSTWriter(sstFile)
-			if err := sst.ClearRawRange(r.Start.Key, r.End.Key); err != nil {
+			if err := sst.ClearRawRange(r.Start, r.End); err != nil {
 				return err
 			}
 
@@ -3800,7 +3800,7 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 				if err != nil {
 					return err
 				}
-				if !valid || r.End.Key.Compare(it.UnsafeKey().Key) <= 0 {
+				if !valid || r.End.Compare(it.UnsafeKey().Key) <= 0 {
 					if err := sst.Finish(); err != nil {
 						return err
 					}
@@ -3828,7 +3828,7 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 			sst := storage.MakeIngestionSSTWriter(sstFile)
 			defer sst.Close()
 			r := rditer.MakeRangeIDLocalKeyRange(rangeID, false /* replicatedOnly */)
-			if err := sst.ClearRawRange(r.Start.Key, r.End.Key); err != nil {
+			if err := sst.ClearRawRange(r.Start, r.End); err != nil {
 				return err
 			}
 			tombstoneKey := keys.RangeTombstoneKey(rangeID)
@@ -3852,7 +3852,7 @@ func TestStoreRangeMergeRaftSnapshot(t *testing.T) {
 			EndKey:   roachpb.RKey(keyEnd),
 		}
 		r := rditer.MakeUserKeyRange(&desc)
-		if err := storage.ClearRangeWithHeuristic(receivingEng, &sst, r.Start.Key, r.End.Key); err != nil {
+		if err := storage.ClearRangeWithHeuristic(receivingEng, &sst, r.Start, r.End); err != nil {
 			return err
 		}
 		err := sst.Finish()

pkg/kv/kvserver/rditer/replica_data_iter.go

@@ -18,10 +18,8 @@ import (
 
 // KeyRange is a helper struct for the ReplicaMVCCDataIterator and
 // ReplicaEngineDataIterator.
-// TODO(sumeer): change these to roachpb.Key since the timestamp is
-// always empty and the code below assumes that fact.
 type KeyRange struct {
-	Start, End storage.MVCCKey
+	Start, End roachpb.Key
 }
 
 // ReplicaMVCCDataIterator provides a complete iteration over MVCC or unversioned
@@ -143,8 +141,8 @@ func MakeRangeIDLocalKeyRange(rangeID roachpb.RangeID, replicatedOnly bool) KeyR
 	}
 	sysRangeIDKey := prefixFn(rangeID)
 	return KeyRange{
-		Start: storage.MakeMVCCMetadataKey(sysRangeIDKey),
-		End:   storage.MakeMVCCMetadataKey(sysRangeIDKey.PrefixEnd()),
+		Start: sysRangeIDKey,
+		End:   sysRangeIDKey.PrefixEnd(),
 	}
 }
 
@@ -154,8 +152,8 @@ func MakeRangeIDLocalKeyRange(rangeID roachpb.RangeID, replicatedOnly bool) KeyR
 // /System), but it actually belongs to [/Table/1, /Table/2).
 func makeRangeLocalKeyRange(d *roachpb.RangeDescriptor) KeyRange {
 	return KeyRange{
-		Start: storage.MakeMVCCMetadataKey(keys.MakeRangeKeyPrefix(d.StartKey)),
-		End:   storage.MakeMVCCMetadataKey(keys.MakeRangeKeyPrefix(d.EndKey)),
+		Start: keys.MakeRangeKeyPrefix(d.StartKey),
+		End:   keys.MakeRangeKeyPrefix(d.EndKey),
 	}
 }
 
@@ -177,12 +175,12 @@ func makeRangeLockTableKeyRanges(d *roachpb.RangeDescriptor) [2]KeyRange {
 	endGlobal, _ := keys.LockTableSingleKey(roachpb.Key(d.EndKey), nil)
 	return [2]KeyRange{
 		{
-			Start: storage.MakeMVCCMetadataKey(startRangeLocal),
-			End:   storage.MakeMVCCMetadataKey(endRangeLocal),
+			Start: startRangeLocal,
+			End:   endRangeLocal,
 		},
 		{
-			Start: storage.MakeMVCCMetadataKey(startGlobal),
-			End:   storage.MakeMVCCMetadataKey(endGlobal),
+			Start: startGlobal,
+			End:   endGlobal,
 		},
 	}
 }
@@ -191,8 +189,8 @@ func makeRangeLockTableKeyRanges(d *roachpb.RangeDescriptor) [2]KeyRange {
 func MakeUserKeyRange(d *roachpb.RangeDescriptor) KeyRange {
 	userKeys := d.KeySpan()
 	return KeyRange{
-		Start: storage.MakeMVCCMetadataKey(userKeys.Key.AsRawKey()),
-		End:   storage.MakeMVCCMetadataKey(userKeys.EndKey.AsRawKey()),
+		Start: userKeys.Key.AsRawKey(),
+		End:   userKeys.EndKey.AsRawKey(),
 	}
 }
 
@@ -240,13 +238,13 @@ func (ri *ReplicaMVCCDataIterator) tryCloseAndCreateIter() {
 		ri.it = ri.reader.NewMVCCIterator(
 			storage.MVCCKeyAndIntentsIterKind,
 			storage.IterOptions{
-				LowerBound: ri.ranges[ri.curIndex].Start.Key,
-				UpperBound: ri.ranges[ri.curIndex].End.Key,
+				LowerBound: ri.ranges[ri.curIndex].Start,
+				UpperBound: ri.ranges[ri.curIndex].End,
 			})
 		if ri.reverse {
-			ri.it.SeekLT(ri.ranges[ri.curIndex].End)
+			ri.it.SeekLT(storage.MakeMVCCMetadataKey(ri.ranges[ri.curIndex].End))
 		} else {
-			ri.it.SeekGE(ri.ranges[ri.curIndex].Start)
+			ri.it.SeekGE(storage.MakeMVCCMetadataKey(ri.ranges[ri.curIndex].Start))
 		}
 		if valid, err := ri.it.Valid(); valid || err != nil {
 			ri.err = err
@@ -356,7 +354,7 @@ func NewReplicaEngineDataIterator(
 // seekStart seeks the iterator to the start of its data range.
 func (ri *ReplicaEngineDataIterator) seekStart() {
 	ri.curIndex = 0
-	ri.valid, ri.err = ri.it.SeekEngineKeyGE(storage.EngineKey{Key: ri.ranges[ri.curIndex].Start.Key})
+	ri.valid, ri.err = ri.it.SeekEngineKeyGE(storage.EngineKey{Key: ri.ranges[ri.curIndex].Start})
 	ri.advance()
 }
 
@@ -383,13 +381,13 @@ func (ri *ReplicaEngineDataIterator) advance() {
 			ri.valid = false
 			return
 		}
-		if k.Key.Compare(ri.ranges[ri.curIndex].End.Key) < 0 {
+		if k.Key.Compare(ri.ranges[ri.curIndex].End) < 0 {
 			return
 		}
 		ri.curIndex++
 		if ri.curIndex < len(ri.ranges) {
 			ri.valid, ri.err = ri.it.SeekEngineKeyGE(
-				storage.EngineKey{Key: ri.ranges[ri.curIndex].Start.Key})
+				storage.EngineKey{Key: ri.ranges[ri.curIndex].Start})
 		} else {
 			ri.valid = false
 			return

pkg/kv/kvserver/rditer/replica_data_iter_test.go

@@ -314,7 +314,7 @@ func TestReplicaDataIterator(t *testing.T) {
 
 func checkOrdering(t *testing.T, ranges []KeyRange) {
 	for i := 1; i < len(ranges); i++ {
-		if ranges[i].Start.Less(ranges[i-1].End) {
+		if ranges[i].Start.Compare(ranges[i-1].End) < 0 {
 			t.Fatalf("ranges need to be ordered and non-overlapping, but %s > %s",
 				ranges[i-1].End, ranges[i].Start)
 		}

pkg/kv/kvserver/rditer/stats.go

@@ -27,11 +27,11 @@ func ComputeStatsForRange(
 	for _, keyRange := range MakeReplicatedKeyRangesExceptLockTable(d) {
 		func() {
 			iter := reader.NewMVCCIterator(storage.MVCCKeyAndIntentsIterKind,
-				storage.IterOptions{UpperBound: keyRange.End.Key})
+				storage.IterOptions{UpperBound: keyRange.End})
 			defer iter.Close()
 
 			var msDelta enginepb.MVCCStats
-			if msDelta, err = iter.ComputeStats(keyRange.Start.Key, keyRange.End.Key, nowNanos); err != nil {
+			if msDelta, err = iter.ComputeStats(keyRange.Start, keyRange.End, nowNanos); err != nil {
 				return
 			}
 			ms.Add(msDelta)

pkg/kv/kvserver/replica_consistency.go

@@ -632,9 +632,9 @@ func (r *Replica) sha512(
 		// using MVCCKeyAndIntentsIterKind and consider all locks here.
 		for _, span := range rditer.MakeReplicatedKeyRangesExceptLockTable(&desc) {
 			iter := snap.NewMVCCIterator(storage.MVCCKeyAndIntentsIterKind,
-				storage.IterOptions{UpperBound: span.End.Key})
+				storage.IterOptions{UpperBound: span.End})
 			spanMS, err := storage.ComputeStatsForRange(
-				iter, span.Start.Key, span.End.Key, 0 /* nowNanos */, visitor,
+				iter, span.Start, span.End, 0 /* nowNanos */, visitor,
 			)
 			iter.Close()
 			if err != nil {

pkg/kv/kvserver/replica_raftstorage.go

@@ -733,7 +733,7 @@ func clearRangeData(
 	}
 
 	for _, keyRange := range keyRanges {
-		if err := clearRangeFn(reader, writer, keyRange.Start.Key, keyRange.End.Key); err != nil {
+		if err := clearRangeFn(reader, writer, keyRange.Start, keyRange.End); err != nil {
 			return err
 		}
 	}
@@ -1067,10 +1067,10 @@ func (r *Replica) clearSubsumedReplicaDiskData(
 		// Compute the total key space covered by the current replica and all
 		// subsumed replicas.
 		for i := range srKeyRanges {
-			if srKeyRanges[i].Start.Key.Compare(totalKeyRanges[i].Start.Key) < 0 {
+			if srKeyRanges[i].Start.Compare(totalKeyRanges[i].Start) < 0 {
 				totalKeyRanges[i].Start = srKeyRanges[i].Start
 			}
-			if srKeyRanges[i].End.Key.Compare(totalKeyRanges[i].End.Key) > 0 {
+			if srKeyRanges[i].End.Compare(totalKeyRanges[i].End) > 0 {
 				totalKeyRanges[i].End = srKeyRanges[i].End
 			}
 		}
@@ -1090,15 +1090,15 @@ func (r *Replica) clearSubsumedReplicaDiskData(
 	// before it completes. It is reasonable for a snapshot for r1 from S3 to
 	// subsume both r1 and r2 in S1.
 	for i := range keyRanges {
-		if totalKeyRanges[i].End.Key.Compare(keyRanges[i].End.Key) > 0 {
+		if totalKeyRanges[i].End.Compare(keyRanges[i].End) > 0 {
 			subsumedReplSSTFile := &storage.MemFile{}
 			subsumedReplSST := storage.MakeIngestionSSTWriter(subsumedReplSSTFile)
 			defer subsumedReplSST.Close()
 			if err := storage.ClearRangeWithHeuristic(
 				r.store.Engine(),
 				&subsumedReplSST,
-				keyRanges[i].End.Key,
-				totalKeyRanges[i].End.Key,
+				keyRanges[i].End,
+				totalKeyRanges[i].End,
 			); err != nil {
 				subsumedReplSST.Close()
 				return err
@@ -1120,7 +1120,7 @@ func (r *Replica) clearSubsumedReplicaDiskData(
 		// Extending to the left implies that either we merged "to the left" (we
 		// don't), or that we're applying a snapshot for another range (we don't do
 		// that either). Something is severely wrong for this to happen.
-		if totalKeyRanges[i].Start.Key.Compare(keyRanges[i].Start.Key) < 0 {
+		if totalKeyRanges[i].Start.Compare(keyRanges[i].Start) < 0 {
 			log.Fatalf(ctx, "subsuming replica to our left; key range: %v; total key range %v",
 				keyRanges[i], totalKeyRanges[i])
 		}

pkg/kv/kvserver/replica_sst_snapshot_storage_test.go

@@ -150,7 +150,7 @@ func TestMultiSSTWriterInitSST(t *testing.T) {
 			sstFile := &storage.MemFile{}
 			sst := storage.MakeIngestionSSTWriter(sstFile)
 			defer sst.Close()
-			err := sst.ClearRawRange(r.Start.Key, r.End.Key)
+			err := sst.ClearRawRange(r.Start, r.End)
 			require.NoError(t, err)
 			err = sst.Finish()
 			require.NoError(t, err)

pkg/kv/kvserver/store_snapshot.go

@@ -160,7 +160,7 @@ func (msstw *multiSSTWriter) initSST(ctx context.Context) error {
 	newSST := storage.MakeIngestionSSTWriter(newSSTFile)
 	msstw.currSST = newSST
 	if err := msstw.currSST.ClearRawRange(
-		msstw.keyRanges[msstw.currRange].Start.Key, msstw.keyRanges[msstw.currRange].End.Key); err != nil {
+		msstw.keyRanges[msstw.currRange].Start, msstw.keyRanges[msstw.currRange].End); err != nil {
 		msstw.currSST.Close()
 		return errors.Wrap(err, "failed to clear range on sst file writer")
 	}
@@ -178,7 +178,7 @@ func (msstw *multiSSTWriter) finalizeSST(ctx context.Context) error {
 }
 
 func (msstw *multiSSTWriter) Put(ctx context.Context, key storage.EngineKey, value []byte) error {
-	for msstw.keyRanges[msstw.currRange].End.Key.Compare(key.Key) <= 0 {
+	for msstw.keyRanges[msstw.currRange].End.Compare(key.Key) <= 0 {
 		// Finish the current SST, write to the file, and move to the next key
 		// range.
 		if err := msstw.finalizeSST(ctx); err != nil {
@@ -188,7 +188,7 @@ func (msstw *multiSSTWriter) Put(ctx context.Context, key storage.EngineKey, val
 			return err
 		}
 	}
-	if msstw.keyRanges[msstw.currRange].Start.Key.Compare(key.Key) > 0 {
+	if msstw.keyRanges[msstw.currRange].Start.Compare(key.Key) > 0 {
 		return errors.AssertionFailedf("client error: expected %s to fall in one of %s", key.Key, msstw.keyRanges)
 	}
 	if err := msstw.currSST.PutEngineKey(key, value); err != nil {