kvserver: use Pebble snapshot for catchup scans

The snapshot is used to create an iterator, which is recreated based on
the storage.snapshot.recreate_iter_duration cluster setting, which
defaults to 20s.

This is mostly plumbing changes, except for catchup_scan.go.

Fixes #133851

Epic: none

Release note (ops change): The cluster setting
storage.snapshot.recreate_iter_duration (default 20s) controls how
frequently a long-lived engine iterator, backed by an engine snapshot,
will be closed and recreated. Currently, it is only used for iterators
used in rangefeed catchup scans.

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

@@ -444,7 +444,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/kvclient/rangefeed/rangefeed_external_test.go

@@ -841,6 +841,11 @@ func TestWithOnDeleteRange(t *testing.T) {
 		// should be visible, because catchup scans do emit tombstones. The range
 		// tombstone should be ordered after the initial point, but before the foo
 		// catchup point, and the previous values should respect the range tombstones.
+		//
+		// NB: When RaceEnabled=true, the range key will be emitted multiple
+		// times, since CatchUpSnapshot.CatchUpScan recreates the iterator at
+		// every step. These duplications are harmless and are de-duped by
+		// testEvents when printing.
 		require.NoError(t, db.Put(ctx, makeKey(srv.Codec(), "covered"), "covered"))
 		require.NoError(t, db.Put(ctx, makeKey(srv.Codec(), "foo"), "covered"))
 		require.NoError(t, db.DelRangeUsingTombstone(ctx, makeKey(srv.Codec(), "a"), makeKey(srv.Codec(), "z")))

pkg/kv/kvserver/rangefeed/BUILD.bazel

@@ -40,6 +40,7 @@ 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,11 +56,11 @@ type bufferedRegistration struct {
 		outputLoopCancelFn func()
 		disconnected       bool
 
-		// catchUpIter is created by replica under raftMu lock when registration is
+		// catchUpSnap is created by replica 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
+		// output loop was not started and catchUpSnap is non-nil at the time that
 		// disconnect is called, it is closed by disconnect.
-		catchUpIter *CatchUpIterator
+		catchUpSnap *CatchUpSnapshot
 	}
 
 	// Number of events that have been written to the buffer but
@@ -74,7 +74,7 @@ func newBufferedRegistration(
 	streamCtx context.Context,
 	span roachpb.Span,
 	startTS hlc.Timestamp,
-	catchUpIter *CatchUpIterator,
+	catchUpSnap *CatchUpSnapshot,
 	withDiff bool,
 	withFiltering bool,
 	withOmitRemote bool,
@@ -100,7 +100,7 @@ func newBufferedRegistration(
 		buf:           make(chan *sharedEvent, bufferSz),
 		blockWhenFull: blockWhenFull,
 	}
-	br.mu.catchUpIter = catchUpIter
+	br.mu.catchUpSnap = catchUpSnap
 	return br
 }
 
@@ -172,9 +172,9 @@ func (br *bufferedRegistration) Disconnect(pErr *kvpb.Error) {
 	br.mu.Lock()
 	defer br.mu.Unlock()
 	if !br.mu.disconnected {
-		if br.mu.catchUpIter != nil {
-			br.mu.catchUpIter.Close()
-			br.mu.catchUpIter = nil
+		if br.mu.catchUpSnap != nil {
+			br.mu.catchUpSnap.Close()
+			br.mu.catchUpSnap = nil
 		}
 		if br.mu.outputLoopCancelFn != nil {
 			br.mu.outputLoopCancelFn()
@@ -303,20 +303,19 @@ 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 catchUpIteratorConstructor, this method
-// is a no-op.
+// If the registration does not have a catchUpSnap, this method is a no-op.
 func (br *bufferedRegistration) maybeRunCatchUpScan(ctx context.Context) error {
-	catchUpIter := br.detachCatchUpIter()
-	if catchUpIter == nil {
+	catchUpSnap := br.detachCatchUpSnap()
+	if catchUpSnap == nil {
 		return nil
 	}
 	start := crtime.NowMono()
 	defer func() {
-		catchUpIter.Close()
+		catchUpSnap.Close()
 		br.metrics.RangeFeedCatchUpScanNanos.Inc(start.Elapsed().Nanoseconds())
 	}()
 
-	return catchUpIter.CatchUpScan(ctx, br.stream.SendUnbuffered, br.withDiff, br.withFiltering, br.withOmitRemote, br.bulkDelivery)
+	return catchUpSnap.CatchUpScan(ctx, br.stream.SendUnbuffered, br.withDiff, br.withFiltering, br.withOmitRemote, br.bulkDelivery)
 }
 
 // Wait for this registration to completely process its internal
@@ -341,13 +340,13 @@ func (br *bufferedRegistration) waitForCaughtUp(ctx context.Context) error {
 	return errors.Errorf("bufferedRegistration %v failed to empty in time", br.Range())
 }
 
-// detachCatchUpIter detaches the catchUpIter that was previously attached.
-func (br *bufferedRegistration) detachCatchUpIter() *CatchUpIterator {
+// detachCatchUpSnap detaches the catchUpSnap that was previously attached.
+func (br *bufferedRegistration) detachCatchUpSnap() *CatchUpSnapshot {
 	br.mu.Lock()
 	defer br.mu.Unlock()
-	catchUpIter := br.mu.catchUpIter
-	br.mu.catchUpIter = nil
-	return catchUpIter
+	catchUpSnap := br.mu.catchUpSnap
+	br.mu.catchUpSnap = nil
+	return catchUpSnap
 }
 
 var overflowLogEvery = log.Every(5 * time.Second)

pkg/kv/kvserver/rangefeed/buffered_sender_test.go

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