kvserver/rangefeed: move from log.KvDistribution to log.KvExec

wenyihu6 · wenyihu6 · commit 6c9feca57c2a · 2025-10-22T11:44:58.000-04:00
diff --git a/pkg/kv/kvserver/rangefeed/budget.go b/pkg/kv/kvserver/rangefeed/budget.go
@@ -248,7 +248,7 @@ type SharedBudgetAllocation struct {
 func (a *SharedBudgetAllocation) Use(ctx context.Context) {
 	if a != nil {
 		if atomic.AddInt32(&a.refCount, 1) == 1 {
-			log.KvDistribution.Fatalf(ctx, "unexpected shared memory allocation usage increase after free")
+			log.KvExec.Fatalf(ctx, "unexpected shared memory allocation usage increase after free")
 		}
 	}
 }
diff --git a/pkg/kv/kvserver/rangefeed/buffered_registration.go b/pkg/kv/kvserver/rangefeed/buffered_registration.go
@@ -196,7 +196,7 @@ func (br *bufferedRegistration) outputLoop(ctx context.Context) error {
 	// If the registration has a catch-up scan, run it.
 	if err := br.maybeRunCatchUpScan(ctx); err != nil {
 		err = errors.Wrap(err, "catch-up scan failed")
-		log.KvDistribution.Errorf(ctx, "%v", err)
+		log.KvExec.Errorf(ctx, "%v", err)
 		return err
 	}
 
@@ -229,7 +229,7 @@ func (br *bufferedRegistration) outputLoop(ctx context.Context) error {
 
 		if overflowed {
 			if wasOverflowedOnFirstIteration && br.shouldLogOverflow(oneCheckpointWithTimestampSent) {
-				log.KvDistribution.Warningf(ctx, "rangefeed %s overflowed during catch up scan from %s (useful checkpoint sent: %v)",
+				log.KvExec.Warningf(ctx, "rangefeed %s overflowed during catch up scan from %s (useful checkpoint sent: %v)",
 					br.span, br.catchUpTimestamp, oneCheckpointWithTimestampSent)
 			}
 
diff --git a/pkg/kv/kvserver/rangefeed/catchup_scan_bench_test.go b/pkg/kv/kvserver/rangefeed/catchup_scan_bench_test.go
@@ -255,13 +255,13 @@ func setupData(
 		absPath = loc
 	}
 	if exists {
-		log.KvDistribution.Infof(ctx, "using existing refresh range benchmark data: %s", absPath)
+		log.KvExec.Infof(ctx, "using existing refresh range benchmark data: %s", absPath)
 		testutils.ReadAllFiles(filepath.Join(loc, "*"))
 		return emk(b, loc, opts.lBaseMaxBytes, opts.rwMode), loc
 	}
 
 	eng := emk(b, loc, opts.lBaseMaxBytes, fs.ReadWrite)
-	log.KvDistribution.Infof(ctx, "creating rangefeed benchmark data: %s", absPath)
+	log.KvExec.Infof(ctx, "creating rangefeed benchmark data: %s", absPath)
 
 	// Generate the same data every time.
 	rng := rand.New(rand.NewSource(1449168817))
@@ -322,7 +322,7 @@ func setupData(
 		// optimizations which change the data size result in the same number of
 		// sstables.
 		if scaled := len(order) / 20; i > 0 && (i%scaled) == 0 {
-			log.KvDistribution.Infof(ctx, "committing (%d/~%d) (%d/%d)", i/scaled, 20, i, len(order))
+			log.KvExec.Infof(ctx, "committing (%d/~%d) (%d/%d)", i/scaled, 20, i, len(order))
 			if err := batch.Commit(false /* sync */); err != nil {
 				b.Fatal(err)
 			}
diff --git a/pkg/kv/kvserver/rangefeed/event_size.go b/pkg/kv/kvserver/rangefeed/event_size.go
@@ -205,7 +205,7 @@ func opsCurrMemUsage(ops opsEvent) int64 {
 		case *enginepb.MVCCAbortTxnOp:
 			currMemUsage += abortTxnOpMemUsage(t.TxnID)
 		default:
-			log.KvDistribution.Fatalf(context.Background(), "unknown logical op %T", t)
+			log.KvExec.Fatalf(context.Background(), "unknown logical op %T", t)
 		}
 	}
 	// For each op, a checkpoint may or may not be published depending on whether
@@ -276,7 +276,7 @@ func MemUsage(e event) int64 {
 		// For sync event, no rangefeed events will be published.
 		return eventOverhead + syncEventOverhead
 	default:
-		log.KvDistribution.Fatalf(context.Background(), "missing event variant: %+v", e)
+		log.KvExec.Fatalf(context.Background(), "missing event variant: %+v", e)
 	}
 	// For empty event, only eventOverhead is accounted.
 	return eventOverhead
diff --git a/pkg/kv/kvserver/rangefeed/registry.go b/pkg/kv/kvserver/rangefeed/registry.go
@@ -191,37 +191,37 @@ func (r *baseRegistration) assertEvent(ctx context.Context, event *kvpb.RangeFee
 	switch t := event.GetValue().(type) {
 	case *kvpb.RangeFeedValue:
 		if t.Key == nil {
-			log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedValue.Key: %v", t)
+			log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedValue.Key: %v", t)
 		}
 		if t.Value.RawBytes == nil {
-			log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedValue.Value.RawBytes: %v", t)
+			log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedValue.Value.RawBytes: %v", t)
 		}
 		if t.Value.Timestamp.IsEmpty() {
-			log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedValue.Value.Timestamp: %v", t)
+			log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedValue.Value.Timestamp: %v", t)
 		}
 	case *kvpb.RangeFeedCheckpoint:
 		if t.Span.Key == nil {
-			log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedCheckpoint.Span.Key: %v", t)
+			log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedCheckpoint.Span.Key: %v", t)
 		}
 	case *kvpb.RangeFeedSSTable:
 		if len(t.Data) == 0 {
-			log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedSSTable.Data: %v", t)
+			log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedSSTable.Data: %v", t)
 		}
 		if len(t.Span.Key) == 0 {
-			log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedSSTable.Span: %v", t)
+			log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedSSTable.Span: %v", t)
 		}
 		if t.WriteTS.IsEmpty() {
-			log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedSSTable.Timestamp: %v", t)
+			log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedSSTable.Timestamp: %v", t)
 		}
 	case *kvpb.RangeFeedDeleteRange:
 		if len(t.Span.Key) == 0 || len(t.Span.EndKey) == 0 {
-			log.KvDistribution.Fatalf(ctx, "unexpected empty key in RangeFeedDeleteRange.Span: %v", t)
+			log.KvExec.Fatalf(ctx, "unexpected empty key in RangeFeedDeleteRange.Span: %v", t)
 		}
 		if t.Timestamp.IsEmpty() {
-			log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedDeleteRange.Timestamp: %v", t)
+			log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedDeleteRange.Timestamp: %v", t)
 		}
 	default:
-		log.KvDistribution.Fatalf(ctx, "unexpected RangeFeedEvent variant: %v", t)
+		log.KvExec.Fatalf(ctx, "unexpected RangeFeedEvent variant: %v", t)
 	}
 }
 
@@ -263,7 +263,7 @@ func (r *baseRegistration) maybeStripEvent(
 			// observed all values up to the checkpoint timestamp over a given
 			// key span if any updates to that span have been filtered out.
 			if !t.Span.Contains(r.span) {
-				log.KvDistribution.Fatalf(ctx, "registration span %v larger than checkpoint span %v", r.span, t.Span)
+				log.KvExec.Fatalf(ctx, "registration span %v larger than checkpoint span %v", r.span, t.Span)
 			}
 			t = copyOnWrite().(*kvpb.RangeFeedCheckpoint)
 			t.Span = r.span
@@ -278,7 +278,7 @@ func (r *baseRegistration) maybeStripEvent(
 		// SSTs are always sent in their entirety, it is up to the caller to
 		// filter out irrelevant entries.
 	default:
-		log.KvDistribution.Fatalf(ctx, "unexpected RangeFeedEvent variant: %v", t)
+		log.KvExec.Fatalf(ctx, "unexpected RangeFeedEvent variant: %v", t)
 	}
 	return ret
 }
@@ -364,7 +364,7 @@ func (reg *registry) Register(ctx context.Context, r registration) {
 	r.setID(reg.nextID())
 	if err := reg.tree.Insert(r, false /* fast */); err != nil {
 		// TODO(erikgrinaker): these errors should arguably be returned.
-		log.KvDistribution.Fatalf(ctx, "%v", err)
+		log.KvExec.Fatalf(ctx, "%v", err)
 	}
 }
 
@@ -400,7 +400,7 @@ func (reg *registry) PublishToOverlapping(
 		// surprising. Revisit this once RangeFeed has more users.
 		minTS = hlc.MaxTimestamp
 	default:
-		log.KvDistribution.Fatalf(ctx, "unexpected RangeFeedEvent variant: %v", t)
+		log.KvExec.Fatalf(ctx, "unexpected RangeFeedEvent variant: %v", t)
 	}
 
 	reg.forOverlappingRegs(ctx, span, func(r registration) (bool, *kvpb.Error) {
@@ -468,13 +468,13 @@ func (reg *registry) remove(ctx context.Context, toDelete []interval.Interface)
 	} else if len(toDelete) == 1 {
 		reg.updateMetricsOnUnregistration(toDelete[0].(registration))
 		if err := reg.tree.Delete(toDelete[0], false /* fast */); err != nil {
-			log.KvDistribution.Fatalf(ctx, "%v", err)
+			log.KvExec.Fatalf(ctx, "%v", err)
 		}
 	} else if len(toDelete) > 1 {
 		for _, i := range toDelete {
 			reg.updateMetricsOnUnregistration(i.(registration))
 			if err := reg.tree.Delete(i, true /* fast */); err != nil {
-				log.KvDistribution.Fatalf(ctx, "%v", err)
+				log.KvExec.Fatalf(ctx, "%v", err)
 			}
 		}
 		reg.tree.AdjustRanges()
diff --git a/pkg/kv/kvserver/rangefeed/resolved_timestamp.go b/pkg/kv/kvserver/rangefeed/resolved_timestamp.go
@@ -224,7 +224,7 @@ func (rts *resolvedTimestamp) consumeLogicalOp(
 		return rts.intentQ.Del(t.TxnID)
 
 	default:
-		log.KvDistribution.Fatalf(ctx, "unknown logical op %T", t)
+		log.KvExec.Fatalf(ctx, "unknown logical op %T", t)
 		return false
 	}
 }
@@ -237,7 +237,7 @@ func (rts *resolvedTimestamp) recompute(ctx context.Context) bool {
 		return false
 	}
 	if rts.closedTS.Less(rts.resolvedTS) {
-		log.KvDistribution.Fatalf(ctx, "closed timestamp below resolved timestamp: %s < %s",
+		log.KvExec.Fatalf(ctx, "closed timestamp below resolved timestamp: %s < %s",
 			rts.closedTS, rts.resolvedTS)
 	}
 	newTS := rts.closedTS
@@ -246,7 +246,7 @@ func (rts *resolvedTimestamp) recompute(ctx context.Context) bool {
 	// timestamps cannot be resolved yet.
 	if txn := rts.intentQ.Oldest(); txn != nil {
 		if txn.timestamp.LessEq(rts.resolvedTS) {
-			log.KvDistribution.Fatalf(ctx, "unresolved txn equal to or below resolved timestamp: %s <= %s",
+			log.KvExec.Fatalf(ctx, "unresolved txn equal to or below resolved timestamp: %s <= %s",
 				txn.timestamp, rts.resolvedTS)
 		}
 		// txn.timestamp cannot be resolved, so the resolved timestamp must be Prev.
@@ -258,7 +258,7 @@ func (rts *resolvedTimestamp) recompute(ctx context.Context) bool {
 	newTS.Logical = 0
 
 	if newTS.Less(rts.resolvedTS) {
-		log.KvDistribution.Fatalf(ctx, "resolved timestamp regression, was %s, recomputed as %s",
+		log.KvExec.Fatalf(ctx, "resolved timestamp regression, was %s, recomputed as %s",
 			rts.resolvedTS, newTS)
 	}
 	return rts.resolvedTS.Forward(newTS)
@@ -271,7 +271,7 @@ func (rts *resolvedTimestamp) assertNoChange(ctx context.Context) {
 	before := rts.resolvedTS
 	changed := rts.recompute(ctx)
 	if changed || before != rts.resolvedTS {
-		log.KvDistribution.Fatalf(ctx, "unexpected resolved timestamp change on recomputation, "+
+		log.KvExec.Fatalf(ctx, "unexpected resolved timestamp change on recomputation, "+
 			"was %s, recomputed as %s", before, rts.resolvedTS)
 	}
 }
@@ -289,9 +289,9 @@ func (rts *resolvedTimestamp) assertOpAboveRTS(
 		err := errors.AssertionFailedf(
 			"resolved timestamp %s equal to or above timestamp of operation %v", rts.resolvedTS, &op)
 		if fatal {
-			log.KvDistribution.Fatalf(ctx, "%v", err)
+			log.KvExec.Fatalf(ctx, "%v", err)
 		} else {
-			log.KvDistribution.Errorf(ctx, "%v", err)
+			log.KvExec.Errorf(ctx, "%v", err)
 		}
 	}
 }
diff --git a/pkg/kv/kvserver/rangefeed/scheduled_processor.go b/pkg/kv/kvserver/rangefeed/scheduled_processor.go
@@ -372,7 +372,7 @@ func (p *ScheduledProcessor) Register(
 			return nil
 		}
 		if !p.Span.AsRawSpanWithNoLocals().Contains(r.Span()) {
-			log.KvDistribution.Fatalf(ctx, "registration %s not in Processor's key range %v", r, p.Span)
+			log.KvExec.Fatalf(ctx, "registration %s not in Processor's key range %v", r, p.Span)
 		}
 
 		// Add the new registration to the registry.
@@ -638,7 +638,7 @@ func runRequest[T interface{}](
 		if buildutil.CrdbTestBuild {
 			select {
 			case <-p.stoppedC:
-				log.KvDistribution.Fatalf(ctx, "processing request on stopped processor")
+				log.KvExec.Fatalf(ctx, "processing request on stopped processor")
 			default:
 			}
 		}
@@ -702,7 +702,7 @@ func (p *ScheduledProcessor) consumeEvent(ctx context.Context, e *event) {
 	case e.sync != nil:
 		if e.sync.testRegCatchupSpan != nil {
 			if err := p.reg.waitForCaughtUp(ctx, *e.sync.testRegCatchupSpan); err != nil {
-				log.KvDistribution.Errorf(
+				log.KvExec.Errorf(
 					ctx,
 					"error waiting for registries to catch up during test, results might be impacted: %s",
 					err,
@@ -711,7 +711,7 @@ func (p *ScheduledProcessor) consumeEvent(ctx context.Context, e *event) {
 		}
 		close(e.sync.c)
 	default:
-		log.KvDistribution.Fatalf(ctx, "missing event variant: %+v", e)
+		log.KvExec.Fatalf(ctx, "missing event variant: %+v", e)
 	}
 }
 
@@ -749,7 +749,7 @@ func (p *ScheduledProcessor) consumeLogicalOps(
 			// No updates to publish.
 
 		default:
-			log.KvDistribution.Fatalf(ctx, "unknown logical op %T", t)
+			log.KvExec.Fatalf(ctx, "unknown logical op %T", t)
 		}
 
 		// Determine whether the operation caused the resolved timestamp to
@@ -793,7 +793,7 @@ func (p *ScheduledProcessor) publishValue(
 	alloc *SharedBudgetAllocation,
 ) {
 	if !p.Span.ContainsKey(roachpb.RKey(key)) {
-		log.KvDistribution.Fatalf(ctx, "key %v not in Processor's key range %v", key, p.Span)
+		log.KvExec.Fatalf(ctx, "key %v not in Processor's key range %v", key, p.Span)
 	}
 
 	var prevVal roachpb.Value
@@ -820,7 +820,7 @@ func (p *ScheduledProcessor) publishDeleteRange(
 ) {
 	span := roachpb.Span{Key: startKey, EndKey: endKey}
 	if !p.Span.ContainsKeyRange(roachpb.RKey(startKey), roachpb.RKey(endKey)) {
-		log.KvDistribution.Fatalf(ctx, "span %s not in Processor's key range %v", span, p.Span)
+		log.KvExec.Fatalf(ctx, "span %s not in Processor's key range %v", span, p.Span)
 	}
 
 	var event kvpb.RangeFeedEvent
@@ -839,10 +839,10 @@ func (p *ScheduledProcessor) publishSSTable(
 	alloc *SharedBudgetAllocation,
 ) {
 	if sstSpan.Equal(roachpb.Span{}) {
-		log.KvDistribution.Fatalf(ctx, "received SSTable without span")
+		log.KvExec.Fatalf(ctx, "received SSTable without span")
 	}
 	if sstWTS.IsEmpty() {
-		log.KvDistribution.Fatalf(ctx, "received SSTable without write timestamp")
+		log.KvExec.Fatalf(ctx, "received SSTable without write timestamp")
 	}
 	p.reg.PublishToOverlapping(ctx, sstSpan, &kvpb.RangeFeedEvent{
 		SST: &kvpb.RangeFeedSSTable{
diff --git a/pkg/kv/kvserver/rangefeed/scheduler.go b/pkg/kv/kvserver/rangefeed/scheduler.go
@@ -493,15 +493,15 @@ func (ss *schedulerShard) processEvents(ctx context.Context) {
 
 		if remaining != 0 && buildutil.CrdbTestBuild {
 			if (remaining^procEventType)&remaining != 0 {
-				log.KvDistribution.Fatalf(ctx,
+				log.KvExec.Fatalf(ctx,
 					"rangefeed processor attempted to reschedule event type %s that was not present in original event set %s",
 					procEventType, remaining)
 			}
 		}
 
 		if e&Stopped != 0 {
 			if remaining != 0 {
-				log.KvDistribution.VWarningf(ctx, 5,
+				log.KvExec.VWarningf(ctx, 5,
 					"rangefeed processor %d didn't process all events on close", entry.id)
 			}
 			// We'll keep Stopped state to avoid calling stopped processor again
diff --git a/pkg/kv/kvserver/rangefeed/stream.go b/pkg/kv/kvserver/rangefeed/stream.go
@@ -86,7 +86,7 @@ func (s *PerRangeEventSink) SendError(err *kvpb.Error) {
 		Error: *transformRangefeedErrToClientError(err),
 	})
 	if ev.Error == nil {
-		log.KvDistribution.Fatalf(context.Background(),
+		log.KvExec.Fatalf(context.Background(),
 			"unexpected: SendWithoutBlocking called with non-error event")
 	}
 	// Silence the error: expected to happen when the buffered sender is closed or
diff --git a/pkg/kv/kvserver/rangefeed/stream_manager.go b/pkg/kv/kvserver/rangefeed/stream_manager.go
@@ -108,7 +108,7 @@ func (sm *StreamManager) NewStream(streamID int64, rangeID roachpb.RangeID) (sin
 	case *UnbufferedSender:
 		return NewPerRangeEventSink(rangeID, streamID, sender)
 	default:
-		log.KvDistribution.Fatalf(context.Background(), "unexpected sender type %T", sm)
+		log.KvExec.Fatalf(context.Background(), "unexpected sender type %T", sm)
 		return nil
 	}
 }
@@ -133,7 +133,7 @@ func (sm *StreamManager) OnError(streamID int64) {
 // DisconnectStream disconnects the stream with the given streamID.
 func (sm *StreamManager) DisconnectStream(streamID int64, err *kvpb.Error) {
 	if err == nil {
-		log.KvDistribution.Fatalf(context.Background(),
+		log.KvExec.Fatalf(context.Background(),
 			"unexpected: DisconnectStream called with nil error")
 		return
 	}
@@ -166,7 +166,7 @@ func (sm *StreamManager) AddStream(streamID int64, d Disconnector) {
 		return
 	}
 	if _, ok := sm.streams.m[streamID]; ok {
-		log.KvDistribution.Fatalf(context.Background(), "stream %d already exists", streamID)
+		log.KvExec.Fatalf(context.Background(), "stream %d already exists", streamID)
 	}
 	sm.streams.m[streamID] = d
 	sm.metrics.ActiveMuxRangeFeed.Inc(1)
@@ -209,7 +209,7 @@ func (sm *StreamManager) Stop(ctx context.Context) {
 	sm.sender.cleanup(ctx)
 	sm.streams.Lock()
 	defer sm.streams.Unlock()
-	log.KvDistribution.VInfof(ctx, 2, "stopping stream manager: disconnecting %d streams", len(sm.streams.m))
+	log.KvExec.VInfof(ctx, 2, "stopping stream manager: disconnecting %d streams", len(sm.streams.m))
 	rangefeedClosedErr := kvpb.NewError(
 		kvpb.NewRangeFeedRetryError(kvpb.RangeFeedRetryError_REASON_RANGEFEED_CLOSED))
 	sm.metrics.ActiveMuxRangeFeed.Dec(int64(len(sm.streams.m)))
@@ -228,7 +228,7 @@ func (sm *StreamManager) Stop(ctx context.Context) {
 // sender.run may also finish without sending anything to the channel.
 func (sm *StreamManager) Error() <-chan error {
 	if sm.errCh == nil {
-		log.KvDistribution.Fatalf(context.Background(), "StreamManager.Error called before StreamManager.Start")
+		log.KvExec.Fatalf(context.Background(), "StreamManager.Error called before StreamManager.Start")
 	}
 	return sm.errCh
 }
diff --git a/pkg/kv/kvserver/rangefeed/task.go b/pkg/kv/kvserver/rangefeed/task.go
@@ -60,7 +60,7 @@ func (s *initResolvedTSScan) Run(ctx context.Context) {
 	if err := s.iterateAndConsume(ctx); err != nil {
 		err = errors.Wrap(err, "initial resolved timestamp scan failed")
 		if ctx.Err() == nil { // cancellation probably caused the error
-			log.KvDistribution.Errorf(ctx, "%v", err)
+			log.KvExec.Errorf(ctx, "%v", err)
 		}
 		s.p.StopWithErr(kvpb.NewError(err))
 	} else {
@@ -249,7 +249,7 @@ func (a *txnPushAttempt) Run(ctx context.Context) {
 	defer a.Cancel()
 	if err := a.pushOldTxns(ctx); err != nil {
 		if ctx.Err() == nil { // cancellation probably caused the error
-			log.KvDistribution.Errorf(ctx, "pushing old intents failed: %v", err)
+			log.KvExec.Errorf(ctx, "pushing old intents failed: %v", err)
 		}
 	}
 }
diff --git a/pkg/kv/kvserver/rangefeed/unbuffered_sender.go b/pkg/kv/kvserver/rangefeed/unbuffered_sender.go
diff --git a/pkg/kv/kvserver/replica_rangefeed.go b/pkg/kv/kvserver/replica_rangefeed.go

Original file line number	Diff line number	Diff line change
`@@ -248,7 +248,7 @@ type SharedBudgetAllocation struct {`
`248`	`248`	`func (a *SharedBudgetAllocation) Use(ctx context.Context) {`
`249`	`249`	`if a != nil {`
`250`	`250`	`if atomic.AddInt32(&a.refCount, 1) == 1 {`
`251`		`- log.KvDistribution.Fatalf(ctx, "unexpected shared memory allocation usage increase after free")`
	`251`	`+ log.KvExec.Fatalf(ctx, "unexpected shared memory allocation usage increase after free")`
`252`	`252`	`}`
`253`	`253`	`}`
`254`	`254`	`}`
Original file line number	Diff line number	Diff line change
`@@ -196,7 +196,7 @@ func (br *bufferedRegistration) outputLoop(ctx context.Context) error {`
`196`	`196`	`// If the registration has a catch-up scan, run it.`
`197`	`197`	`if err := br.maybeRunCatchUpScan(ctx); err != nil {`
`198`	`198`	`err = errors.Wrap(err, "catch-up scan failed")`
`199`		`- log.KvDistribution.Errorf(ctx, "%v", err)`
	`199`	`+ log.KvExec.Errorf(ctx, "%v", err)`
`200`	`200`	`return err`
`201`	`201`	`}`
`202`	`202`
`@@ -229,7 +229,7 @@ func (br *bufferedRegistration) outputLoop(ctx context.Context) error {`
`229`	`229`
`230`	`230`	`if overflowed {`
`231`	`231`	`if wasOverflowedOnFirstIteration && br.shouldLogOverflow(oneCheckpointWithTimestampSent) {`
`232`		`- log.KvDistribution.Warningf(ctx, "rangefeed %s overflowed during catch up scan from %s (useful checkpoint sent: %v)",`
	`232`	`+ log.KvExec.Warningf(ctx, "rangefeed %s overflowed during catch up scan from %s (useful checkpoint sent: %v)",`
`233`	`233`	`br.span, br.catchUpTimestamp, oneCheckpointWithTimestampSent)`
`234`	`234`	`}`
`235`	`235`
Original file line number	Diff line number	Diff line change
`@@ -205,7 +205,7 @@ func opsCurrMemUsage(ops opsEvent) int64 {`
`205`	`205`	`case *enginepb.MVCCAbortTxnOp:`
`206`	`206`	`currMemUsage += abortTxnOpMemUsage(t.TxnID)`
`207`	`207`	`default:`
`208`		`- log.KvDistribution.Fatalf(context.Background(), "unknown logical op %T", t)`
	`208`	`+ log.KvExec.Fatalf(context.Background(), "unknown logical op %T", t)`
`209`	`209`	`}`
`210`	`210`	`}`
`211`	`211`	`// For each op, a checkpoint may or may not be published depending on whether`
`@@ -276,7 +276,7 @@ func MemUsage(e event) int64 {`
`276`	`276`	`// For sync event, no rangefeed events will be published.`
`277`	`277`	`return eventOverhead + syncEventOverhead`
`278`	`278`	`default:`
`279`		`- log.KvDistribution.Fatalf(context.Background(), "missing event variant: %+v", e)`
	`279`	`+ log.KvExec.Fatalf(context.Background(), "missing event variant: %+v", e)`
`280`	`280`	`}`
`281`	`281`	`// For empty event, only eventOverhead is accounted.`
`282`	`282`	`return eventOverhead`
Original file line number	Diff line number	Diff line change
`@@ -191,37 +191,37 @@ func (r baseRegistration) assertEvent(ctx context.Context, event kvpb.RangeFee`
`191`	`191`	`switch t := event.GetValue().(type) {`
`192`	`192`	`case *kvpb.RangeFeedValue:`
`193`	`193`	`if t.Key == nil {`
`194`		`- log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedValue.Key: %v", t)`
	`194`	`+ log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedValue.Key: %v", t)`
`195`	`195`	`}`
`196`	`196`	`if t.Value.RawBytes == nil {`
`197`		`- log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedValue.Value.RawBytes: %v", t)`
	`197`	`+ log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedValue.Value.RawBytes: %v", t)`
`198`	`198`	`}`
`199`	`199`	`if t.Value.Timestamp.IsEmpty() {`
`200`		`- log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedValue.Value.Timestamp: %v", t)`
	`200`	`+ log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedValue.Value.Timestamp: %v", t)`
`201`	`201`	`}`
`202`	`202`	`case *kvpb.RangeFeedCheckpoint:`
`203`	`203`	`if t.Span.Key == nil {`
`204`		`- log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedCheckpoint.Span.Key: %v", t)`
	`204`	`+ log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedCheckpoint.Span.Key: %v", t)`
`205`	`205`	`}`
`206`	`206`	`case *kvpb.RangeFeedSSTable:`
`207`	`207`	`if len(t.Data) == 0 {`
`208`		`- log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedSSTable.Data: %v", t)`
	`208`	`+ log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedSSTable.Data: %v", t)`
`209`	`209`	`}`
`210`	`210`	`if len(t.Span.Key) == 0 {`
`211`		`- log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedSSTable.Span: %v", t)`
	`211`	`+ log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedSSTable.Span: %v", t)`
`212`	`212`	`}`
`213`	`213`	`if t.WriteTS.IsEmpty() {`
`214`		`- log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedSSTable.Timestamp: %v", t)`
	`214`	`+ log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedSSTable.Timestamp: %v", t)`
`215`	`215`	`}`
`216`	`216`	`case *kvpb.RangeFeedDeleteRange:`
`217`	`217`	`if len(t.Span.Key) == 0 \|\| len(t.Span.EndKey) == 0 {`
`218`		`- log.KvDistribution.Fatalf(ctx, "unexpected empty key in RangeFeedDeleteRange.Span: %v", t)`
	`218`	`+ log.KvExec.Fatalf(ctx, "unexpected empty key in RangeFeedDeleteRange.Span: %v", t)`
`219`	`219`	`}`
`220`	`220`	`if t.Timestamp.IsEmpty() {`
`221`		`- log.KvDistribution.Fatalf(ctx, "unexpected empty RangeFeedDeleteRange.Timestamp: %v", t)`
	`221`	`+ log.KvExec.Fatalf(ctx, "unexpected empty RangeFeedDeleteRange.Timestamp: %v", t)`
`222`	`222`	`}`
`223`	`223`	`default:`
`224`		`- log.KvDistribution.Fatalf(ctx, "unexpected RangeFeedEvent variant: %v", t)`
	`224`	`+ log.KvExec.Fatalf(ctx, "unexpected RangeFeedEvent variant: %v", t)`
`225`	`225`	`}`
`226`	`226`	`}`
`227`	`227`
`@@ -263,7 +263,7 @@ func (r *baseRegistration) maybeStripEvent(`
`263`	`263`	`// observed all values up to the checkpoint timestamp over a given`
`264`	`264`	`// key span if any updates to that span have been filtered out.`
`265`	`265`	`if !t.Span.Contains(r.span) {`
`266`		`- log.KvDistribution.Fatalf(ctx, "registration span %v larger than checkpoint span %v", r.span, t.Span)`
	`266`	`+ log.KvExec.Fatalf(ctx, "registration span %v larger than checkpoint span %v", r.span, t.Span)`
`267`	`267`	`}`
`268`	`268`	`t = copyOnWrite().(*kvpb.RangeFeedCheckpoint)`
`269`	`269`	`t.Span = r.span`
`@@ -278,7 +278,7 @@ func (r *baseRegistration) maybeStripEvent(`
`278`	`278`	`// SSTs are always sent in their entirety, it is up to the caller to`
`279`	`279`	`// filter out irrelevant entries.`
`280`	`280`	`default:`
`281`		`- log.KvDistribution.Fatalf(ctx, "unexpected RangeFeedEvent variant: %v", t)`
	`281`	`+ log.KvExec.Fatalf(ctx, "unexpected RangeFeedEvent variant: %v", t)`
`282`	`282`	`}`
`283`	`283`	`return ret`
`284`	`284`	`}`
`@@ -364,7 +364,7 @@ func (reg *registry) Register(ctx context.Context, r registration) {`
`364`	`364`	`r.setID(reg.nextID())`
`365`	`365`	`if err := reg.tree.Insert(r, false /* fast */); err != nil {`
`366`	`366`	`// TODO(erikgrinaker): these errors should arguably be returned.`
`367`		`- log.KvDistribution.Fatalf(ctx, "%v", err)`
	`367`	`+ log.KvExec.Fatalf(ctx, "%v", err)`
`368`	`368`	`}`
`369`	`369`	`}`
`370`	`370`
`@@ -400,7 +400,7 @@ func (reg *registry) PublishToOverlapping(`
`400`	`400`	`// surprising. Revisit this once RangeFeed has more users.`
`401`	`401`	`minTS = hlc.MaxTimestamp`
`402`	`402`	`default:`
`403`		`- log.KvDistribution.Fatalf(ctx, "unexpected RangeFeedEvent variant: %v", t)`
	`403`	`+ log.KvExec.Fatalf(ctx, "unexpected RangeFeedEvent variant: %v", t)`
`404`	`404`	`}`
`405`	`405`
`406`	`406`	`reg.forOverlappingRegs(ctx, span, func(r registration) (bool, *kvpb.Error) {`
`@@ -468,13 +468,13 @@ func (reg *registry) remove(ctx context.Context, toDelete []interval.Interface)`
`468`	`468`	`} else if len(toDelete) == 1 {`
`469`	`469`	`reg.updateMetricsOnUnregistration(toDelete[0].(registration))`
`470`	`470`	`if err := reg.tree.Delete(toDelete[0], false /* fast */); err != nil {`
`471`		`- log.KvDistribution.Fatalf(ctx, "%v", err)`
	`471`	`+ log.KvExec.Fatalf(ctx, "%v", err)`
`472`	`472`	`}`
`473`	`473`	`} else if len(toDelete) > 1 {`
`474`	`474`	`for _, i := range toDelete {`
`475`	`475`	`reg.updateMetricsOnUnregistration(i.(registration))`
`476`	`476`	`if err := reg.tree.Delete(i, true /* fast */); err != nil {`
`477`		`- log.KvDistribution.Fatalf(ctx, "%v", err)`
	`477`	`+ log.KvExec.Fatalf(ctx, "%v", err)`
`478`	`478`	`}`
`479`	`479`	`}`
`480`	`480`	`reg.tree.AdjustRanges()`
Original file line number	Diff line number	Diff line change
`@@ -224,7 +224,7 @@ func (rts *resolvedTimestamp) consumeLogicalOp(`
`224`	`224`	`return rts.intentQ.Del(t.TxnID)`
`225`	`225`
`226`	`226`	`default:`
`227`		`- log.KvDistribution.Fatalf(ctx, "unknown logical op %T", t)`
	`227`	`+ log.KvExec.Fatalf(ctx, "unknown logical op %T", t)`
`228`	`228`	`return false`
`229`	`229`	`}`
`230`	`230`	`}`
`@@ -237,7 +237,7 @@ func (rts *resolvedTimestamp) recompute(ctx context.Context) bool {`
`237`	`237`	`return false`
`238`	`238`	`}`
`239`	`239`	`if rts.closedTS.Less(rts.resolvedTS) {`
`240`		`- log.KvDistribution.Fatalf(ctx, "closed timestamp below resolved timestamp: %s < %s",`
	`240`	`+ log.KvExec.Fatalf(ctx, "closed timestamp below resolved timestamp: %s < %s",`
`241`	`241`	`rts.closedTS, rts.resolvedTS)`
`242`	`242`	`}`
`243`	`243`	`newTS := rts.closedTS`
`@@ -246,7 +246,7 @@ func (rts *resolvedTimestamp) recompute(ctx context.Context) bool {`
`246`	`246`	`// timestamps cannot be resolved yet.`
`247`	`247`	`if txn := rts.intentQ.Oldest(); txn != nil {`
`248`	`248`	`if txn.timestamp.LessEq(rts.resolvedTS) {`
`249`		`- log.KvDistribution.Fatalf(ctx, "unresolved txn equal to or below resolved timestamp: %s <= %s",`
	`249`	`+ log.KvExec.Fatalf(ctx, "unresolved txn equal to or below resolved timestamp: %s <= %s",`
`250`	`250`	`txn.timestamp, rts.resolvedTS)`
`251`	`251`	`}`
`252`	`252`	`// txn.timestamp cannot be resolved, so the resolved timestamp must be Prev.`
`@@ -258,7 +258,7 @@ func (rts *resolvedTimestamp) recompute(ctx context.Context) bool {`
`258`	`258`	`newTS.Logical = 0`
`259`	`259`
`260`	`260`	`if newTS.Less(rts.resolvedTS) {`
`261`		`- log.KvDistribution.Fatalf(ctx, "resolved timestamp regression, was %s, recomputed as %s",`
	`261`	`+ log.KvExec.Fatalf(ctx, "resolved timestamp regression, was %s, recomputed as %s",`
`262`	`262`	`rts.resolvedTS, newTS)`
`263`	`263`	`}`
`264`	`264`	`return rts.resolvedTS.Forward(newTS)`
`@@ -271,7 +271,7 @@ func (rts *resolvedTimestamp) assertNoChange(ctx context.Context) {`
`271`	`271`	`before := rts.resolvedTS`
`272`	`272`	`changed := rts.recompute(ctx)`
`273`	`273`	`if changed \|\| before != rts.resolvedTS {`
`274`		`- log.KvDistribution.Fatalf(ctx, "unexpected resolved timestamp change on recomputation, "+`
	`274`	`+ log.KvExec.Fatalf(ctx, "unexpected resolved timestamp change on recomputation, "+`
`275`	`275`	`"was %s, recomputed as %s", before, rts.resolvedTS)`
`276`	`276`	`}`
`277`	`277`	`}`
`@@ -289,9 +289,9 @@ func (rts *resolvedTimestamp) assertOpAboveRTS(`
`289`	`289`	`err := errors.AssertionFailedf(`
`290`	`290`	`"resolved timestamp %s equal to or above timestamp of operation %v", rts.resolvedTS, &op)`
`291`	`291`	`if fatal {`
`292`		`- log.KvDistribution.Fatalf(ctx, "%v", err)`
	`292`	`+ log.KvExec.Fatalf(ctx, "%v", err)`
`293`	`293`	`} else {`
`294`		`- log.KvDistribution.Errorf(ctx, "%v", err)`
	`294`	`+ log.KvExec.Errorf(ctx, "%v", err)`
`295`	`295`	`}`
`296`	`296`	`}`
`297`	`297`	`}`
Original file line number	Diff line number	Diff line change
`@@ -372,7 +372,7 @@ func (p *ScheduledProcessor) Register(`
`372`	`372`	`return nil`
`373`	`373`	`}`
`374`	`374`	`if !p.Span.AsRawSpanWithNoLocals().Contains(r.Span()) {`
`375`		`- log.KvDistribution.Fatalf(ctx, "registration %s not in Processor's key range %v", r, p.Span)`
	`375`	`+ log.KvExec.Fatalf(ctx, "registration %s not in Processor's key range %v", r, p.Span)`
`376`	`376`	`}`
`377`	`377`
`378`	`378`	`// Add the new registration to the registry.`
`@@ -638,7 +638,7 @@ func runRequest[T interface{}](`
`638`	`638`	`if buildutil.CrdbTestBuild {`
`639`	`639`	`select {`
`640`	`640`	`case <-p.stoppedC:`
`641`		`- log.KvDistribution.Fatalf(ctx, "processing request on stopped processor")`
	`641`	`+ log.KvExec.Fatalf(ctx, "processing request on stopped processor")`
`642`	`642`	`default:`
`643`	`643`	`}`
`644`	`644`	`}`
`@@ -702,7 +702,7 @@ func (p ScheduledProcessor) consumeEvent(ctx context.Context, e event) {`
`702`	`702`	`case e.sync != nil:`
`703`	`703`	`if e.sync.testRegCatchupSpan != nil {`
`704`	`704`	`if err := p.reg.waitForCaughtUp(ctx, *e.sync.testRegCatchupSpan); err != nil {`
`705`		`- log.KvDistribution.Errorf(`
	`705`	`+ log.KvExec.Errorf(`
`706`	`706`	`ctx,`
`707`	`707`	`"error waiting for registries to catch up during test, results might be impacted: %s",`
`708`	`708`	`err,`
`@@ -711,7 +711,7 @@ func (p ScheduledProcessor) consumeEvent(ctx context.Context, e event) {`
`711`	`711`	`}`
`712`	`712`	`close(e.sync.c)`
`713`	`713`	`default:`
`714`		`- log.KvDistribution.Fatalf(ctx, "missing event variant: %+v", e)`
	`714`	`+ log.KvExec.Fatalf(ctx, "missing event variant: %+v", e)`
`715`	`715`	`}`
`716`	`716`	`}`
`717`	`717`
`@@ -749,7 +749,7 @@ func (p *ScheduledProcessor) consumeLogicalOps(`
`749`	`749`	`// No updates to publish.`
`750`	`750`
`751`	`751`	`default:`
`752`		`- log.KvDistribution.Fatalf(ctx, "unknown logical op %T", t)`
	`752`	`+ log.KvExec.Fatalf(ctx, "unknown logical op %T", t)`
`753`	`753`	`}`
`754`	`754`
`755`	`755`	`// Determine whether the operation caused the resolved timestamp to`
`@@ -793,7 +793,7 @@ func (p *ScheduledProcessor) publishValue(`
`793`	`793`	`alloc *SharedBudgetAllocation,`
`794`	`794`	`) {`
`795`	`795`	`if !p.Span.ContainsKey(roachpb.RKey(key)) {`
`796`		`- log.KvDistribution.Fatalf(ctx, "key %v not in Processor's key range %v", key, p.Span)`
	`796`	`+ log.KvExec.Fatalf(ctx, "key %v not in Processor's key range %v", key, p.Span)`
`797`	`797`	`}`
`798`	`798`
`799`	`799`	`var prevVal roachpb.Value`
`@@ -820,7 +820,7 @@ func (p *ScheduledProcessor) publishDeleteRange(`
`820`	`820`	`) {`
`821`	`821`	`span := roachpb.Span{Key: startKey, EndKey: endKey}`
`822`	`822`	`if !p.Span.ContainsKeyRange(roachpb.RKey(startKey), roachpb.RKey(endKey)) {`
`823`		`- log.KvDistribution.Fatalf(ctx, "span %s not in Processor's key range %v", span, p.Span)`
	`823`	`+ log.KvExec.Fatalf(ctx, "span %s not in Processor's key range %v", span, p.Span)`
`824`	`824`	`}`
`825`	`825`
`826`	`826`	`var event kvpb.RangeFeedEvent`
`@@ -839,10 +839,10 @@ func (p *ScheduledProcessor) publishSSTable(`
`839`	`839`	`alloc *SharedBudgetAllocation,`
`840`	`840`	`) {`
`841`	`841`	`if sstSpan.Equal(roachpb.Span{}) {`
`842`		`- log.KvDistribution.Fatalf(ctx, "received SSTable without span")`
	`842`	`+ log.KvExec.Fatalf(ctx, "received SSTable without span")`
`843`	`843`	`}`
`844`	`844`	`if sstWTS.IsEmpty() {`
`845`		`- log.KvDistribution.Fatalf(ctx, "received SSTable without write timestamp")`
	`845`	`+ log.KvExec.Fatalf(ctx, "received SSTable without write timestamp")`
`846`	`846`	`}`
`847`	`847`	`p.reg.PublishToOverlapping(ctx, sstSpan, &kvpb.RangeFeedEvent{`
`848`	`848`	`SST: &kvpb.RangeFeedSSTable{`
Original file line number	Diff line number	Diff line change
`@@ -493,15 +493,15 @@ func (ss *schedulerShard) processEvents(ctx context.Context) {`
`493`	`493`
`494`	`494`	`if remaining != 0 && buildutil.CrdbTestBuild {`
`495`	`495`	`if (remaining^procEventType)&remaining != 0 {`
`496`		`- log.KvDistribution.Fatalf(ctx,`
	`496`	`+ log.KvExec.Fatalf(ctx,`
`497`	`497`	`"rangefeed processor attempted to reschedule event type %s that was not present in original event set %s",`
`498`	`498`	`procEventType, remaining)`
`499`	`499`	`}`
`500`	`500`	`}`
`501`	`501`
`502`	`502`	`if e&Stopped != 0 {`
`503`	`503`	`if remaining != 0 {`
`504`		`- log.KvDistribution.VWarningf(ctx, 5,`
	`504`	`+ log.KvExec.VWarningf(ctx, 5,`
`505`	`505`	`"rangefeed processor %d didn't process all events on close", entry.id)`
`506`	`506`	`}`
`507`	`507`	`// We'll keep Stopped state to avoid calling stopped processor again`
Original file line number	Diff line number	Diff line change
`@@ -86,7 +86,7 @@ func (s PerRangeEventSink) SendError(err kvpb.Error) {`
`86`	`86`	`Error: *transformRangefeedErrToClientError(err),`
`87`	`87`	`})`
`88`	`88`	`if ev.Error == nil {`
`89`		`- log.KvDistribution.Fatalf(context.Background(),`
	`89`	`+ log.KvExec.Fatalf(context.Background(),`
`90`	`90`	`"unexpected: SendWithoutBlocking called with non-error event")`
`91`	`91`	`}`
`92`	`92`	`// Silence the error: expected to happen when the buffered sender is closed or`