Merge cockroachdb#97957

97957: kvcoord: MuxRangeFeed client uses 1 go routine per node r=miretskiy a=miretskiy

Rewrite MuxRangeFeed client to use 1 Go routine per node,
instead of 1 Go routine per range.

Prior to this change, MuxRangeFeed client was structured
so that it was entirely compatible with the execution
model of the regular range feed.  As a result,
1 Go routine was used per range.  This rewrite replaces
old implementation with an almost clean slate implementation
which uses 1 Go routine per node.

Where possible, relatively small and targetted modifications
to the rangefeed library were made to extract common methods
(such as range splitting).

The reduction in the number of Go routines created by rangefeed
has direct impact on the cluster performance, and most importantly
SQL latency.  This is mostly due to the fact that with this PR,
the number of Go routines started by MuxRangeFeed is down to
2 per range (on the rangefeed server side) vs 5 for the regular
rangefeed.  When running changefeeds against tables with
10s-100s of thousands of ranges, this significant difference
in the Go routine count has direct impact on Go scheduler latency,
the number of runnable Go routines, and ultimately, on the SQL
latency.

Epic: none

Release note (enterprise change) : MuxRangeFeed client (enabled via
`changefeed.mux_rangefeed.enabled` setting) is more efficient
when running against large scale workloads.

Co-authored-by: Yevgeniy Miretskiy <yevgeniy@cockroachlabs.com>

pkg/ccl/changefeedccl/kvfeed/kv_feed_test.go

@@ -398,7 +398,6 @@ type rawEventFeed []kvpb.RangeFeedEvent
 func (f rawEventFeed) run(
 	ctx context.Context,
 	spans []kvcoord.SpanTimePair,
-	withDiff bool,
 	eventC chan<- kvcoord.RangeFeedMessage,
 	opts ...kvcoord.RangeFeedOption,
 ) error {

pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go

@@ -37,7 +37,6 @@ type rangeFeedConfig struct {
 type rangefeedFactory func(
 	ctx context.Context,
 	spans []kvcoord.SpanTimePair,
-	withDiff bool,
 	eventC chan<- kvcoord.RangeFeedMessage,
 	opts ...kvcoord.RangeFeedOption,
 ) error
@@ -79,9 +78,12 @@ func (p rangefeedFactory) Run(ctx context.Context, sink kvevent.Writer, cfg rang
 	if cfg.UseMux {
 		rfOpts = append(rfOpts, kvcoord.WithMuxRangeFeed())
 	}
+	if cfg.WithDiff {
+		rfOpts = append(rfOpts, kvcoord.WithDiff())
+	}
 
 	g.GoCtx(func(ctx context.Context) error {
-		return p(ctx, cfg.Spans, cfg.WithDiff, feed.eventC, rfOpts...)
+		return p(ctx, cfg.Spans, feed.eventC, rfOpts...)
 	})
 	return g.Wait()
 }

pkg/kv/kvclient/kvcoord/BUILD.bazel

@@ -69,6 +69,7 @@ go_library(
         "//pkg/util/ctxgroup",
         "//pkg/util/envutil",
         "//pkg/util/errorutil/unimplemented",
+        "//pkg/util/future",
         "//pkg/util/grpcutil",
         "//pkg/util/hlc",
         "//pkg/util/iterutil",