kvserver: remove unnecessary comparison in catchup scan

Iterators already respect the upper bound we set at construction time
making this comparison redundant. Removing this comparison provides a
nice performance win:

```
name                                                               old time/op    new time/op    delta
CatchupScan/linear-keys/useTBI=false/withDiff=true/perc=99.00-16      261ms ± 2%     249ms ± 4%  -4.90%  (p=0.000 n=24+25)
CatchupScan/linear-keys/useTBI=false/withDiff=false/perc=99.00-16     226ms ± 3%     214ms ± 4%  -5.55%  (p=0.000 n=23+25)
```

Release note: None

Release justification: Performance improvement for catchup scans helps
address the cause of recent high-priority customer incidents.

pkg/kv/kvserver/rangefeed/catchup_scan.go

@@ -129,7 +129,7 @@ func (i *CatchUpIterator) CatchUpScan(
 	for {
 		if ok, err := i.Valid(); err != nil {
 			return err
-		} else if !ok || !i.UnsafeKey().Less(endKey) {
+		} else if !ok {
 			break
 		}
 

pkg/kv/kvserver/rangefeed/processor.go

@@ -185,6 +185,8 @@ type IteratorConstructor func() storage.SimpleMVCCIterator
 // CatchUpIteratorConstructor is used to construct an iterator that
 // can be used for catchup-scans. It should be called from underneath
 // a stopper task to ensure that the engine has not been closed.
+//
+// The constructed iterator must have an UpperBound set.
 type CatchUpIteratorConstructor func() *CatchUpIterator
 
 // Start launches a goroutine to process rangefeed events and send them to

pkg/kv/kvserver/rangefeed/processor_test.go

@@ -563,7 +563,7 @@ func TestProcessorInitializeResolvedTimestamp(t *testing.T) {
 		makeIntent("z", txn2, "txnKey2", 21),
 		makeProvisionalKV("z", "txnKey2", 21),
 		makeKV("z", "val11", 4),
-	})
+	}, nil)
 	rtsIter.block = make(chan struct{})
 
 	p, stopper := newTestProcessor(rtsIter)

pkg/kv/kvserver/rangefeed/registry_test.go

@@ -163,7 +163,7 @@ func TestRegistrationBasic(t *testing.T) {
 		makeInline("ba", "val2"),
 		makeKV("bc", "val3", 11),
 		makeKV("bd", "val4", 9),
-	}), false)
+	}, nil), false)
 	catchupReg.publish(ev1)
 	catchupReg.publish(ev2)
 	require.Equal(t, len(catchupReg.buf), 2)
@@ -266,7 +266,7 @@ func TestRegistrationCatchUpScan(t *testing.T) {
 		makeIntent("z", txn2, "txnKeyZ", 21),
 		makeProvisionalKV("z", "txnKeyZ", 21),
 		makeKV("z", "valZ1", 4),
-	})
+	}, roachpb.Key("w"))
 	r := newTestRegistration(roachpb.Span{
 		Key:    roachpb.Key("d"),
 		EndKey: roachpb.Key("w"),

pkg/kv/kvserver/rangefeed/task_test.go

@@ -71,16 +71,17 @@ func makeIntent(key string, txnID uuid.UUID, txnKey string, txnTS int64) storage
 }
 
 type testIterator struct {
-	kvs []storage.MVCCKeyValue
-	cur int
+	kvs        []storage.MVCCKeyValue
+	cur        int
+	upperBound roachpb.Key
 
 	closed bool
 	err    error
 	block  chan struct{}
 	done   chan struct{}
 }
 
-func newTestIterator(kvs []storage.MVCCKeyValue) *testIterator {
+func newTestIterator(kvs []storage.MVCCKeyValue, upperBound roachpb.Key) *testIterator {
 	// Ensure that the key-values are sorted.
 	if !sort.SliceIsSorted(kvs, func(i, j int) bool {
 		return kvs[i].Key.Less(kvs[j].Key)
@@ -115,9 +116,10 @@ func newTestIterator(kvs []storage.MVCCKeyValue) *testIterator {
 	}
 
 	return &testIterator{
-		kvs:  kvs,
-		cur:  -1,
-		done: make(chan struct{}),
+		kvs:        kvs,
+		cur:        -1,
+		done:       make(chan struct{}),
+		upperBound: upperBound,
 	}
 }
 
@@ -153,6 +155,9 @@ func (s *testIterator) Valid() (bool, error) {
 	if s.cur == -1 || s.cur >= len(s.kvs) {
 		return false, nil
 	}
+	if s.upperBound != nil && !s.curKV().Key.Less(storage.MVCCKey{Key: s.upperBound}) {
+		return false, nil
+	}
 	return true, nil
 }
 
@@ -223,7 +228,7 @@ func TestInitResolvedTSScan(t *testing.T) {
 		makeIntent("z", txn2, "txnKey2", 21),
 		makeProvisionalKV("z", "txnKey2", 21),
 		makeKV("z", "val11", 4),
-	})
+	}, nil)
 
 	initScan := newInitResolvedTSScan(&p, iter)
 	initScan.Run(context.Background())