changefeedccl: make bulk delivery of rangefeed events optional

This is a temporary opt-out until we can properly test the performance
impact of bulk delivery.

Epic: none

Release note (general change): The changefeed bulk
delivery setting was made optional.

Author: asg0451

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -597,6 +597,7 @@ func (ca *changeAggregator) makeKVFeedCfg(
 		WithDiff:             filters.WithDiff,
 		WithFiltering:        filters.WithFiltering,
 		WithFrontierQuantize: changefeedbase.Quantize.Get(&cfg.Settings.SV),
+		WithBulkDelivery:     changefeedbase.BulkDelivery.Get(&cfg.Settings.SV),
 		NeedsInitialScan:     needsInitialScan,
 		SchemaChangeEvents:   schemaChange.EventClass,
 		SchemaChangePolicy:   schemaChange.Policy,

pkg/ccl/changefeedccl/changefeedbase/settings.go

@@ -224,6 +224,14 @@ var PerTableProtectedTimestamps = settings.RegisterBoolSetting(
 		"if false, uses a single protected timestamp record for all tables",
 	metamorphic.ConstantWithTestBool("changefeed.protect_timestamp.per_table.enabled", false))
 
+// BulkDelivery enables bulk delivery of rangefeed events, which can improve performance during catchup scans.
+var BulkDelivery = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"changefeed.bulk_delivery.enabled",
+	"if true, rangefeed events are delivered in bulk during catchup scans; "+
+		"if false, rangefeed events are delivered individually",
+	metamorphic.ConstantWithTestBool("changefeed.bulk_delivery.enabled", true))
+
 // MaxProtectedTimestampAge controls the frequency of protected timestamp record updates
 var MaxProtectedTimestampAge = settings.RegisterDurationSetting(
 	settings.ApplicationLevel,

pkg/ccl/changefeedccl/kvfeed/kv_feed.go

@@ -94,6 +94,11 @@ type Config struct {
 	// enables filtering out any transactional writes with that flag set to true.
 	WithFiltering bool
 
+	// WithBulkDelivery is propagated via the RangefeedRequest to the rangefeed
+	// server, where if true, the server will deliver rangefeed events in bulk
+	// during catchup scans.
+	WithBulkDelivery bool
+
 	// WithFrontierQuantize specifies the resolved timestamp quantization
 	// granularity. If non-zero, resolved timestamps from rangefeed checkpoint
 	// events will be rounded down to the nearest multiple of the quantization
@@ -135,7 +140,7 @@ func Run(ctx context.Context, cfg Config) error {
 	f := newKVFeed(
 		cfg.Writer, cfg.Spans,
 		cfg.SchemaChangeEvents, cfg.SchemaChangePolicy,
-		cfg.NeedsInitialScan, cfg.WithDiff, cfg.WithFiltering,
+		cfg.NeedsInitialScan, cfg.WithDiff, cfg.WithFiltering, cfg.WithBulkDelivery,
 		cfg.WithFrontierQuantize,
 		cfg.ConsumerID,
 		cfg.InitialHighWater, cfg.InitialSpanTimePairs, cfg.EndTime,
@@ -260,6 +265,7 @@ type kvFeed struct {
 	withDiff             bool
 	withFiltering        bool
 	withInitialBackfill  bool
+	withBulkDelivery     bool
 	consumerID           int64
 	initialHighWater     hlc.Timestamp
 	initialSpanTimePairs []kvcoord.SpanTimePair
@@ -289,7 +295,7 @@ func newKVFeed(
 	spans []roachpb.Span,
 	schemaChangeEvents changefeedbase.SchemaChangeEventClass,
 	schemaChangePolicy changefeedbase.SchemaChangePolicy,
-	withInitialBackfill, withDiff, withFiltering bool,
+	withInitialBackfill, withDiff, withFiltering, withBulkDelivery bool,
 	withFrontierQuantize time.Duration,
 	consumerID int64,
 	initialHighWater hlc.Timestamp,
@@ -312,6 +318,7 @@ func newKVFeed(
 		withDiff:             withDiff,
 		withFiltering:        withFiltering,
 		withFrontierQuantize: withFrontierQuantize,
+		withBulkDelivery:     withBulkDelivery,
 		consumerID:           consumerID,
 		initialHighWater:     initialHighWater,
 		endTime:              endTime,
@@ -615,6 +622,7 @@ func (f *kvFeed) runUntilTableEvent(ctx context.Context, resumeFrontier span.Fro
 		WithDiff:             f.withDiff,
 		WithFiltering:        f.withFiltering,
 		WithFrontierQuantize: f.withFrontierQuantize,
+		WithBulkDelivery:     f.withBulkDelivery,
 		ConsumerID:           f.consumerID,
 		Knobs:                f.knobs,
 		Timers:               f.timers,

pkg/ccl/changefeedccl/kvfeed/kv_feed_test.go

@@ -148,7 +148,7 @@ func TestKVFeed(t *testing.T) {
 		st := timers.New(time.Minute).GetOrCreateScopedTimers("")
 		f := newKVFeed(buf, tc.spans,
 			tc.schemaChangeEvents, tc.schemaChangePolicy,
-			tc.needsInitialScan, tc.withDiff, true /* withFiltering */, tc.withFrontierQuantize,
+			tc.needsInitialScan, tc.withDiff, true /* withFiltering */, changefeedbase.BulkDelivery.Get(&settings.SV), tc.withFrontierQuantize,
 			0, /* consumerID */
 			tc.initialHighWater, tc.initialSpanTimePairs, tc.endTime,
 			codec,

pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go

@@ -34,6 +34,7 @@ type rangeFeedConfig struct {
 	WithDiff             bool
 	WithFiltering        bool
 	WithFrontierQuantize time.Duration
+	WithBulkDelivery     bool
 	ConsumerID           int64
 	RangeObserver        kvcoord.RangeObserver
 	Knobs                TestingKnobs
@@ -97,7 +98,10 @@ func (p rangefeedFactory) Run(ctx context.Context, sink kvevent.Writer, cfg rang
 	// Bulk delivery is an optimization that decreases rangefeed overhead during
 	// catchup scans. It results in the emission of BulkEvents instead of
 	// individual events where possible.
-	rfOpts := []kvcoord.RangeFeedOption{kvcoord.WithBulkDelivery()}
+	var rfOpts []kvcoord.RangeFeedOption
+	if cfg.WithBulkDelivery {
+		rfOpts = append(rfOpts, kvcoord.WithBulkDelivery())
+	}
 	if cfg.WithDiff {
 		rfOpts = append(rfOpts, kvcoord.WithDiff())
 	}