Merge #150371

150371: kvserver,rangefeed: enable bulk value delivery r=dt a=dt

When many values are available at the same time, such as during catch-up scans, transmitting and delivering each as a separate event incurs significant overhead and contention. Instead, we introduce a new event type which is a BulkEvent that consists of many events. During catch-up scans we can buffer up value events and then emit many of them in a single bulk event.

Not all callers can handle these new bulk events so the rangefeed server must be asked to emit them, however a client using the kvclient rangefeed wrapper can benefit from them even if it does not directly support them as the client wrapper will automatically enable bulk delivery and then unwrap them for delivery to the caller individually.

Release note: none.
Epic: none.

Co-authored-by: David Taylor <tinystatemachine@gmail.com>

Author: craig[bot]

pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go

@@ -178,6 +178,9 @@ func (p *rangefeed) addEventsToBuffer(ctx context.Context) error {
 				// For now, we just error on SST ingestion, since we currently don't
 				// expect SST ingestion into spans with active changefeeds.
 				return errors.Errorf("unexpected SST ingestion: %v", t)
+			case *kvpb.RangeFeedBulkEvents:
+				// Should be disabled so this is unreachable.
+				return errors.Errorf("unexpected bulk delivery: %v", t)
 
 			case *kvpb.RangeFeedDeleteRange:
 				// For now, we just ignore on MVCC range tombstones. These are currently

pkg/kv/kvclient/kvcoord/dist_sender_mux_rangefeed.go

@@ -283,7 +283,7 @@ func (s *activeMuxRangeFeed) start(ctx context.Context, m *rangefeedMuxer) error
 
 		for !s.transport.IsExhausted() {
 			args := makeRangeFeedRequest(
-				s.Span, s.token.Desc().RangeID, m.cfg.overSystemTable, s.startAfter, m.cfg.withDiff, m.cfg.withFiltering, m.cfg.withMatchingOriginIDs, m.cfg.consumerID)
+				s.Span, s.token.Desc().RangeID, m.cfg.overSystemTable, s.startAfter, m.cfg.withDiff, m.cfg.withFiltering, m.cfg.withMatchingOriginIDs, m.cfg.consumerID, m.cfg.bulkDelivery)
 			args.Replica = s.transport.NextReplica()
 			args.StreamID = streamID
 			s.ReplicaDescriptor = args.Replica

pkg/kv/kvclient/kvcoord/dist_sender_rangefeed.go

@@ -69,6 +69,7 @@ type rangeFeedConfig struct {
 	withMatchingOriginIDs []uint32
 	rangeObserver         RangeObserver
 	consumerID            int64
+	bulkDelivery          bool
 
 	knobs struct {
 		// onRangefeedEvent invoked on each rangefeed event.
@@ -118,6 +119,12 @@ func WithFiltering() RangeFeedOption {
 	})
 }
 
+func WithBulkDelivery() RangeFeedOption {
+	return optionFunc(func(c *rangeFeedConfig) {
+		c.bulkDelivery = true
+	})
+}
+
 // WithMatchingOriginIDs opts the rangefeed into emitting events originally written by
 // clusters with the assoicated origin IDs during logical data replication.
 func WithMatchingOriginIDs(originIDs ...uint32) RangeFeedOption {
@@ -633,6 +640,7 @@ func makeRangeFeedRequest(
 	withFiltering bool,
 	withMatchingOriginIDs []uint32,
 	consumerID int64,
+	withBulkDelivery bool,
 ) kvpb.RangeFeedRequest {
 	admissionPri := admissionpb.BulkNormalPri
 	if isSystemRange {
@@ -648,6 +656,7 @@ func makeRangeFeedRequest(
 		WithDiff:              withDiff,
 		WithFiltering:         withFiltering,
 		WithMatchingOriginIDs: withMatchingOriginIDs,
+		WithBulkDelivery:      withBulkDelivery,
 		AdmissionHeader: kvpb.AdmissionHeader{
 			// NB: AdmissionHeader is used only at the start of the range feed
 			// stream since the initial catch-up scan is expensive.

pkg/kv/kvclient/rangefeed/rangefeed.go

@@ -332,6 +332,12 @@ func (f *RangeFeed) run(ctx context.Context, frontier span.Frontier, resumeWithF
 	eventCh := make(chan kvcoord.RangeFeedMessage)
 
 	var rangefeedOpts []kvcoord.RangeFeedOption
+	// We can unconditionally enable bulk-delivery from the server at least as far
+	// as to this client; if an onValues is configured we can also bulk-process
+	// values, but even if it isn't we know how to unwrap a bulk delivery and pass
+	// each event to the caller's individual event handlers.
+	rangefeedOpts = append(rangefeedOpts, kvcoord.WithBulkDelivery())
+
 	if f.scanConfig.overSystemTable {
 		rangefeedOpts = append(rangefeedOpts, kvcoord.WithSystemTablePriority())
 	}
@@ -427,54 +433,98 @@ func (f *RangeFeed) processEvents(
 	for {
 		select {
 		case ev := <-eventCh:
-			switch {
-			case ev.Val != nil:
-				f.onValue(ctx, ev.Val)
-			case ev.Checkpoint != nil:
-				ts := ev.Checkpoint.ResolvedTS
-				if f.frontierQuantize != 0 {
-					ts.Logical = 0
-					ts.WallTime -= ts.WallTime % int64(f.frontierQuantize)
-				}
-				advanced, err := frontier.Forward(ev.Checkpoint.Span, ts)
-				if err != nil {
-					return err
-				}
-				if f.onCheckpoint != nil {
-					f.onCheckpoint(ctx, ev.Checkpoint)
-				}
-				if advanced && f.onFrontierAdvance != nil {
-					f.onFrontierAdvance(ctx, frontier.Frontier())
-				}
-				if f.frontierVisitor != nil {
-					f.frontierVisitor(ctx, advanced, frontier)
-				}
-			case ev.SST != nil:
-				if f.onSSTable == nil {
-					return errors.AssertionFailedf(
-						"received unexpected rangefeed SST event with no OnSSTable handler")
-				}
-				f.onSSTable(ctx, ev.SST, ev.RegisteredSpan)
-			case ev.DeleteRange != nil:
-				if f.onDeleteRange == nil {
-					if f.knobs != nil && f.knobs.IgnoreOnDeleteRangeError {
-						continue
-					}
-					return errors.AssertionFailedf(
-						"received unexpected rangefeed DeleteRange event with no OnDeleteRange handler: %s", ev)
-				}
-				f.onDeleteRange(ctx, ev.DeleteRange)
-			case ev.Metadata != nil:
-				if f.onMetadata == nil {
-					return errors.AssertionFailedf("received unexpected metadata event with no OnMetadata handler")
-				}
-				f.onMetadata(ctx, ev.Metadata)
-			case ev.Error != nil:
-				// Intentionally do nothing, we'll get an error returned from the
-				// call to RangeFeed.
+			if err := f.processEvent(ctx, frontier, ev.RangeFeedEvent, ev.RegisteredSpan); err != nil {
+				return err
 			}
 		case <-ctx.Done():
 			return ctx.Err()
 		}
 	}
 }
+
+func (f *RangeFeed) processEvent(
+	ctx context.Context, frontier span.Frontier, ev *kvpb.RangeFeedEvent, registeredSpan roachpb.Span,
+) error {
+	switch {
+	case ev.Val != nil:
+		f.onValue(ctx, ev.Val)
+	case ev.Checkpoint != nil:
+		ts := ev.Checkpoint.ResolvedTS
+		if f.frontierQuantize != 0 {
+			ts.Logical = 0
+			ts.WallTime -= ts.WallTime % int64(f.frontierQuantize)
+		}
+		advanced, err := frontier.Forward(ev.Checkpoint.Span, ts)
+		if err != nil {
+			return err
+		}
+		if f.onCheckpoint != nil {
+			f.onCheckpoint(ctx, ev.Checkpoint)
+		}
+		if advanced && f.onFrontierAdvance != nil {
+			f.onFrontierAdvance(ctx, frontier.Frontier())
+		}
+		if f.frontierVisitor != nil {
+			f.frontierVisitor(ctx, advanced, frontier)
+		}
+	case ev.SST != nil:
+		if f.onSSTable == nil {
+			return errors.AssertionFailedf(
+				"received unexpected rangefeed SST event with no OnSSTable handler")
+		}
+		f.onSSTable(ctx, ev.SST, registeredSpan)
+	case ev.DeleteRange != nil:
+		if f.onDeleteRange == nil {
+			if f.knobs != nil && f.knobs.IgnoreOnDeleteRangeError {
+				return nil
+			}
+			return errors.AssertionFailedf(
+				"received unexpected rangefeed DeleteRange event with no OnDeleteRange handler: %s", ev)
+		}
+		f.onDeleteRange(ctx, ev.DeleteRange)
+	case ev.Metadata != nil:
+		if f.onMetadata == nil {
+			return errors.AssertionFailedf("received unexpected metadata event with no OnMetadata handler")
+		}
+		f.onMetadata(ctx, ev.Metadata)
+	case ev.Error != nil:
+		// Intentionally do nothing, we'll get an error returned from the
+		// call to RangeFeed.
+	case ev.BulkEvents != nil:
+		if f.onValues != nil {
+			// We can optimistically assume the bulk event consists of all value
+			// events, and allocate a buffer for them to be passed to onValues. In the
+			// rare case we hit a non-value event (it would have to be a range key as
+			// only a catch-up scan currently produces bulk events), we will throw out
+			// this buffer and any events we might have copied to it so far and just
+			// fallback to to processing each event, but this should be so uncommon it
+			// is not worth worrying about the potential wasted work.
+			allValues := true
+			buf := make([]kv.KeyValue, len(ev.BulkEvents.Events))
+			for i := range ev.BulkEvents.Events {
+				if ev.BulkEvents.Events[i].Val != nil {
+					buf[i] = kv.KeyValue{
+						Key:   ev.BulkEvents.Events[i].Val.Key,
+						Value: &ev.BulkEvents.Events[i].Val.Value,
+					}
+				} else {
+					allValues = false
+					break
+				}
+			}
+			if allValues {
+				f.onValues(ctx, buf)
+				return nil
+			}
+		}
+		// Either the bulk event contains non-value events or a onValues handler is
+		// not configured, so process each event individually.
+		for _, e := range ev.BulkEvents.Events {
+			if err := f.processEvent(ctx, frontier, e, registeredSpan); err != nil {
+				return err
+			}
+		}
+
+	}
+	return nil
+}

pkg/kv/kvpb/api.go

@@ -2160,6 +2160,9 @@ func (e *RangeFeedEvent) ShallowCopy() *RangeFeedEvent {
 	case *RangeFeedError:
 		cpyErr := *t
 		cpy.MustSetValue(&cpyErr)
+	case *RangeFeedBulkEvents:
+		cpyVals := *t
+		cpy.MustSetValue(&cpyVals)
 	default:
 		panic(fmt.Sprintf("unexpected RangeFeedEvent variant: %v", t))
 	}

pkg/kv/kvpb/api.proto

@@ -3280,7 +3280,11 @@ message RangeFeedRequest {
   // ConsumerID is set by the caller to identify itself.
   int64 consumer_id = 9 [(gogoproto.customname) = "ConsumerID"];
 
-  // NextID = 10;
+  // WithBulkDelivery requests the stream include "bulk" events which wrap many
+  // events in a single event to reduce overhead, e.g. during scans.
+  bool with_bulk_delivery = 10;
+
+  // NextID = 11;
 }
 
 // RangeFeedValue is a variant of RangeFeedEvent that represents an update to
@@ -3296,6 +3300,14 @@ message RangeFeedValue {
   Value prev_value = 3 [(gogoproto.nullable) = false];
 }
 
+// RangeFeedBulkEvents is a variant of RangeFeedEvent that represetns a
+// collection of many RangeFeedEvent delivered as a single event rather than 
+// individually for reduced delivery overhead when many events are available, 
+// such as during catch-up scans.
+message RangeFeedBulkEvents {
+  repeated RangeFeedEvent events = 1;
+}
+
 // RangeFeedCheckpoint is a variant of RangeFeedEvent that represents the
 // promise that no more RangeFeedValue events with keys in the specified span
 // with timestamps less than or equal to the specified resolved timestamp will
@@ -3365,6 +3377,7 @@ message RangeFeedEvent {
   RangeFeedSSTable     sst          = 4 [(gogoproto.customname) = "SST"];
   RangeFeedDeleteRange delete_range = 5;
   RangeFeedMetadata    metadata     = 6;
+  RangeFeedBulkEvents  bulk_events  = 7;
 }
 
 // MuxRangeFeedEvent is a response generated by MuxRangeFeed RPC.  It tags

pkg/kv/kvserver/rangefeed/bench_test.go

@@ -104,7 +104,7 @@ func runBenchmarkRangefeed(b *testing.B, opts benchmarkRangefeedOpts) {
 		const withFiltering = false
 		streams[i] = &noopStream{ctx: ctx, done: make(chan *kvpb.Error, 1)}
 		ok, _, _ := p.Register(ctx, span, hlc.MinTimestamp, nil,
-			withDiff, withFiltering, false, /* withOmitRemote */
+			withDiff, withFiltering, false /* withOmitRemote */, false, /* withBulkDelivery */
 			streams[i])
 		require.True(b, ok)
 	}

pkg/kv/kvserver/rangefeed/buffered_registration.go

@@ -78,6 +78,7 @@ func newBufferedRegistration(
 	withDiff bool,
 	withFiltering bool,
 	withOmitRemote bool,
+	withBulkDelivery bool,
 	bufferSz int,
 	blockWhenFull bool,
 	metrics *Metrics,
@@ -93,6 +94,7 @@ func newBufferedRegistration(
 			withFiltering:          withFiltering,
 			withOmitRemote:         withOmitRemote,
 			removeRegFromProcessor: removeRegFromProcessor,
+			bulkDelivery:           withBulkDelivery,
 		},
 		metrics:       metrics,
 		stream:        stream,
@@ -308,7 +310,7 @@ func (br *bufferedRegistration) maybeRunCatchUpScan(ctx context.Context) error {
 		br.metrics.RangeFeedCatchUpScanNanos.Inc(timeutil.Since(start).Nanoseconds())
 	}()
 
-	return catchUpIter.CatchUpScan(ctx, br.stream.SendUnbuffered, br.withDiff, br.withFiltering, br.withOmitRemote)
+	return catchUpIter.CatchUpScan(ctx, br.stream.SendUnbuffered, br.withDiff, br.withFiltering, br.withOmitRemote, br.bulkDelivery)
 }
 
 // Wait for this registration to completely process its internal

pkg/kv/kvserver/rangefeed/catchup_scan.go

@@ -119,6 +119,8 @@ func (i *CatchUpIterator) Close() {
 // TODO(ssd): Clarify memory ownership. Currently, the memory backing
 // the RangeFeedEvents isn't modified by the caller after this
 // returns. However, we may revist this in #69596.
+// TODO(dt): Does this really need to be a pointer to a struct containing all
+// pointers? can we pass by value instead?
 type outputEventFn func(e *kvpb.RangeFeedEvent) error
 
 // CatchUpScan iterates over all changes in the configured key/time span, and
@@ -137,10 +139,11 @@ type outputEventFn func(e *kvpb.RangeFeedEvent) error
 // to SimpleMVCCIterator to replace the context.
 func (i *CatchUpIterator) CatchUpScan(
 	ctx context.Context,
-	outputFn outputEventFn,
+	emitFn outputEventFn,
 	withDiff bool,
 	withFiltering bool,
 	withOmitRemote bool,
+	bulkDelivery bool,
 ) error {
 	var a bufalloc.ByteAllocator
 	// MVCCIterator will encounter historical values for each key in
@@ -150,6 +153,26 @@ func (i *CatchUpIterator) CatchUpScan(
 	// as we fill in previous values.
 	reorderBuf := make([]kvpb.RangeFeedEvent, 0, 5)
 
+	outputFn := emitFn
+	var emitBufSize int
+	var emitBuf []*kvpb.RangeFeedEvent
+
+	if bulkDelivery {
+		outputFn = func(event *kvpb.RangeFeedEvent) error {
+			emitBuf = append(emitBuf, event)
+			emitBufSize += event.Size()
+			// If there are ~2MB of buffered events, flush them.
+			if emitBufSize >= 2<<20 {
+				if err := emitFn(&kvpb.RangeFeedEvent{BulkEvents: &kvpb.RangeFeedBulkEvents{Events: emitBuf}}); err != nil {
+					return err
+				}
+				emitBuf = make([]*kvpb.RangeFeedEvent, 0, len(emitBuf))
+				emitBufSize = 0
+			}
+			return nil
+		}
+	}
+
 	outputEvents := func() error {
 		for i := len(reorderBuf) - 1; i >= 0; i-- {
 			e := reorderBuf[i]
@@ -383,5 +406,16 @@ func (i *CatchUpIterator) CatchUpScan(
 	}
 
 	// Output events for the last key encountered.
-	return outputEvents()
+	if err := outputEvents(); err != nil {
+		return err
+	}
+	// If bulk delivery has buffered anything for emission, flush it.
+	if len(emitBuf) > 0 {
+		// Flush any remaining buffered events.
+		if err := emitFn(&kvpb.RangeFeedEvent{BulkEvents: &kvpb.RangeFeedBulkEvents{Events: emitBuf}}); err != nil {
+			return err
+		}
+	}
+
+	return nil
 }

pkg/kv/kvserver/rangefeed/catchup_scan_bench_test.go

@@ -54,7 +54,7 @@ func runCatchUpBenchmark(b *testing.B, emk engineMaker, opts benchOptions) (numE
 			err = iter.CatchUpScan(ctx, func(*kvpb.RangeFeedEvent) error {
 				counter++
 				return nil
-			}, opts.withDiff, false /* withFiltering */, false /* withOmitRemote */)
+			}, opts.withDiff, false /* withFiltering */, false /* withOmitRemote */, false)
 			if err != nil {
 				b.Fatalf("failed catchUp scan: %+v", err)
 			}