kvserver: Revert "kvserver: use Pebble snapshot for catchup scans"

This reverts commit 6e59022.
Due to test failures #154566,
#154575.

Epic: none

Release note: None

Author: sumeerbhola

pkg/backup/backupsink/file_sst_sink.go

@@ -361,7 +361,7 @@ func (s *FileSSTSink) copyPointKeys(ctx context.Context, dataSST []byte) (roachp
 
 	empty := true
 	for iter.SeekGE(storage.MVCCKey{Key: keys.MinKey}); ; iter.Next() {
-		if _, err := s.pacer.Pace(ctx); err != nil {
+		if err := s.pacer.Pace(ctx); err != nil {
 			return nil, err
 		}
 		if valid, err := iter.Valid(); !valid || err != nil {

pkg/backup/compaction_processor.go

@@ -422,7 +422,7 @@ func compactSpanEntry(
 	scratch = append(scratch, prefix...)
 	iter := sstIter.iter
 	for iter.SeekGE(trimmedStart); ; iter.NextKey() {
-		if _, err := pacer.Pace(ctx); err != nil {
+		if err := pacer.Pace(ctx); err != nil {
 			return err
 		}
 		var key storage.MVCCKey

pkg/ccl/changefeedccl/batching_sink.go

@@ -476,7 +476,7 @@ func (s *batchingSink) runBatchingWorker(ctx context.Context) {
 			// TODO(yevgeniy): rework this function: this function should simply
 			// return an error, and not rely on "handleError".
 			// It's hard to reason about this functions correctness otherwise.
-			_, _ = s.pacer.Pace(ctx)
+			_ = s.pacer.Pace(ctx)
 
 			switch r := req.(type) {
 			case *rowEvent:

pkg/ccl/changefeedccl/event_processing.go

@@ -343,7 +343,7 @@ func (c *kvEventToRowConsumer) ConsumeEvent(ctx context.Context, ev kvevent.Even
 	// Request CPU time to use for event consumption, block if this time is
 	// unavailable. If there is unused CPU time left from the last call to
 	// Pace, then use that time instead of blocking.
-	if _, err := c.pacer.Pace(ctx); err != nil {
+	if err := c.pacer.Pace(ctx); err != nil {
 		return err
 	}
 

pkg/kv/bulk/sst_batcher.go

@@ -445,7 +445,7 @@ func (b *SSTBatcher) AddMVCCKeyLDR(ctx context.Context, key storage.MVCCKey, val
 // Keys must be added in order.
 func (b *SSTBatcher) AddMVCCKey(ctx context.Context, key storage.MVCCKey, value []byte) error {
 	// Pace based on admission control before adding the key.
-	if _, err := b.pacer.Pace(ctx); err != nil {
+	if err := b.pacer.Pace(ctx); err != nil {
 		return err
 	}
 

pkg/kv/kvserver/rangefeed/BUILD.bazel

@@ -39,7 +39,6 @@ go_library(
         "//pkg/storage",
         "//pkg/storage/enginepb",
         "//pkg/storage/fs",
-        "//pkg/util",
         "//pkg/util/admission",
         "//pkg/util/bufalloc",
         "//pkg/util/buildutil",

pkg/kv/kvserver/rangefeed/buffered_registration.go

@@ -56,12 +56,11 @@ type bufferedRegistration struct {
 		outputLoopCancelFn func()
 		disconnected       bool
 
-		// catchUpSnap is created by a replica under the raftMu lock when a
-		// registration is created. It is detached by the output loop for
-		// processing and closed. If the output loop was not started and
-		// catchUpSnap is non-nil at the time that disconnect is called, it is
-		// closed by disconnect.
-		catchUpSnap *CatchUpSnapshot
+		// catchUpIter is created by replcia under raftMu lock when registration is
+		// created. It is detached by output loop for processing and closed.
+		// If output loop was not started and catchUpIter is non-nil at the time
+		// that disconnect is called, it is closed by disconnect.
+		catchUpIter *CatchUpIterator
 	}
 
 	// Number of events that have been written to the buffer but
@@ -75,7 +74,7 @@ func newBufferedRegistration(
 	streamCtx context.Context,
 	span roachpb.Span,
 	startTS hlc.Timestamp,
-	catchUpSnap *CatchUpSnapshot,
+	catchUpIter *CatchUpIterator,
 	withDiff bool,
 	withFiltering bool,
 	withOmitRemote bool,
@@ -102,7 +101,7 @@ func newBufferedRegistration(
 		buf:           make(chan *sharedEvent, bufferSz),
 		blockWhenFull: blockWhenFull,
 	}
-	br.mu.catchUpSnap = catchUpSnap
+	br.mu.catchUpIter = catchUpIter
 	return br
 }
 
@@ -167,9 +166,9 @@ func (br *bufferedRegistration) Disconnect(pErr *kvpb.Error) {
 	br.mu.Lock()
 	defer br.mu.Unlock()
 	if !br.mu.disconnected {
-		if br.mu.catchUpSnap != nil {
-			br.mu.catchUpSnap.Close()
-			br.mu.catchUpSnap = nil
+		if br.mu.catchUpIter != nil {
+			br.mu.catchUpIter.Close()
+			br.mu.catchUpIter = nil
 		}
 		if br.mu.outputLoopCancelFn != nil {
 			br.mu.outputLoopCancelFn()
@@ -298,19 +297,20 @@ func (br *bufferedRegistration) drainAllocations(ctx context.Context) {
 // This uses the iterator provided when the registration was originally created;
 // after the scan completes, the iterator will be closed.
 //
-// If the registration does not have a catchUpSnap, this method is a no-op.
+// If the registration does not have a catchUpIteratorConstructor, this method
+// is a no-op.
 func (br *bufferedRegistration) maybeRunCatchUpScan(ctx context.Context) error {
-	catchUpSnap := br.detachCatchUpSnap()
-	if catchUpSnap == nil {
+	catchUpIter := br.detachCatchUpIter()
+	if catchUpIter == nil {
 		return nil
 	}
 	start := crtime.NowMono()
 	defer func() {
-		catchUpSnap.Close()
+		catchUpIter.Close()
 		br.metrics.RangeFeedCatchUpScanNanos.Inc(start.Elapsed().Nanoseconds())
 	}()
 
-	return catchUpSnap.CatchUpScan(ctx, br.stream.SendUnbuffered, br.withDiff, br.withFiltering, br.withOmitRemote, br.bulkDelivery)
+	return catchUpIter.CatchUpScan(ctx, br.stream.SendUnbuffered, br.withDiff, br.withFiltering, br.withOmitRemote, br.bulkDelivery)
 }
 
 // Wait for this registration to completely process its internal
@@ -335,13 +335,13 @@ func (br *bufferedRegistration) waitForCaughtUp(ctx context.Context) error {
 	return errors.Errorf("bufferedRegistration %v failed to empty in time", br.Range())
 }
 
-// detachCatchUpSnap detaches the catchUpSnap that was previously attached.
-func (br *bufferedRegistration) detachCatchUpSnap() *CatchUpSnapshot {
+// detachCatchUpIter detaches the catchUpIter that was previously attached.
+func (br *bufferedRegistration) detachCatchUpIter() *CatchUpIterator {
 	br.mu.Lock()
 	defer br.mu.Unlock()
-	catchUpSnap := br.mu.catchUpSnap
-	br.mu.catchUpSnap = nil
-	return catchUpSnap
+	catchUpIter := br.mu.catchUpIter
+	br.mu.catchUpIter = nil
+	return catchUpIter
 }
 
 var overflowLogEvery = log.Every(5 * time.Second)

pkg/kv/kvserver/rangefeed/buffered_sender_test.go

@@ -258,7 +258,7 @@ func TestBufferedSenderOnStreamShutdown(t *testing.T) {
 	}
 
 	// Add our stream to the stream manager.
-	registered, d, _ := p.Register(ctx, h.span, hlc.Timestamp{}, nil, /* catchUpSnap */
+	registered, d, _ := p.Register(ctx, h.span, hlc.Timestamp{}, nil, /* catchUpIter */
 		false /* withDiff */, false /* withFiltering */, false /* withOmitRemote */, noBulkDelivery,
 		sm.NewStream(streamID, 1 /*rangeID*/))
 	require.True(t, registered)