changefeedccl: have aggregators send their frontiers regularly

This patch updates the aggregators to send their frontiers regularly
every `min_checkpoint_frequency` (limited by average time to flush)
to support periodic span frontier persistence.

Release note: None

Author: andyyang890

pkg/ccl/changefeedccl/BUILD.bazel

@@ -203,6 +203,7 @@ go_test(
         "alter_changefeed_test.go",
         "changefeed_dist_test.go",
         "changefeed_job_info_test.go",
+        "changefeed_processors_test.go",
         "changefeed_progress_test.go",
         "changefeed_stmt_test.go",
         "changefeed_test.go",

pkg/ccl/changefeedccl/alter_changefeed_test.go

@@ -1524,13 +1524,7 @@ func TestAlterChangefeedAddTargetsDuringBackfill(t *testing.T) {
 			defer rndMu.Unlock()
 
 			if r.Span.Equal(fooTableSpan) {
-				// Do not emit resolved events for the entire table span.
-				// We "simulate" large table by splitting single table span into many parts, so
-				// we want to resolve those sub-spans instead of the entire table span.
-				// However, we have to emit something -- otherwise the entire changefeed
-				// machine would not work.
-				r.Span.EndKey = fooTableSpan.Key.Next()
-				return false, nil
+				return true, nil
 			}
 			if haveGaps {
 				return rndMu.rnd.Intn(10) > 7, nil

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -106,9 +106,11 @@ type changeAggregator struct {
 	// eventConsumer consumes the event.
 	eventConsumer eventConsumer
 
-	nextHighWaterFlush time.Time     // next time high watermark may be flushed.
-	flushFrequency     time.Duration // how often high watermark can be checkpointed.
-	lastSpanFlush      time.Time     // last time expensive, span based checkpoint was written.
+	flushFrequency time.Duration // how often high watermark can be checkpointed.
+
+	// frontierFlushLimiter is a rate limiter for flushing the span frontier
+	// to the coordinator.
+	frontierFlushLimiter *saveRateLimiter
 
 	// frontier keeps track of resolved timestamps for spans along with schema change
 	// boundary information.
@@ -281,6 +283,22 @@ func newChangeAggregatorProcessor(
 		ca.flushFrequency = changefeedbase.DefaultMinCheckpointFrequency
 	}
 
+	ca.frontierFlushLimiter, err = newSaveRateLimiter(saveRateConfig{
+		name: "frontier",
+		intervalName: func() redact.SafeValue {
+			return redact.SafeString(changefeedbase.OptMinCheckpointFrequency)
+		},
+		interval: func() time.Duration {
+			return ca.flushFrequency
+		},
+		jitter: func() float64 {
+			return aggregatorFlushJitter.Get(&ca.FlowCtx.Cfg.Settings.SV)
+		},
+	}, timeutil.DefaultTimeSource{})
+	if err != nil {
+		return nil, err
+	}
+
 	return ca, nil
 }
 
@@ -461,9 +479,6 @@ func (ca *changeAggregator) Start(ctx context.Context) {
 
 	// Init heartbeat timer.
 	ca.lastPush = timeutil.Now()
-
-	// Generate expensive checkpoint only after we ran for a while.
-	ca.lastSpanFlush = timeutil.Now()
 }
 
 func (ca *changeAggregator) startKVFeed(
@@ -730,18 +745,6 @@ var aggregatorFlushJitter = settings.RegisterFloatSetting(
 	settings.WithPublic,
 )
 
-func nextFlushWithJitter(s timeutil.TimeSource, d time.Duration, j float64) (time.Time, error) {
-	if j < 0 || d < 0 {
-		return s.Now(), errors.AssertionFailedf("invalid jitter value: %f, duration: %s", j, d)
-	}
-	maxJitter := int64(j * float64(d))
-	if maxJitter == 0 {
-		return s.Now().Add(d), nil
-	}
-	nextFlush := d + time.Duration(rand.Int63n(maxJitter))
-	return s.Now().Add(nextFlush), nil
-}
-
 // Next is part of the RowSource interface.
 func (ca *changeAggregator) Next() (rowenc.EncDatumRow, *execinfrapb.ProducerMetadata) {
 	shouldEmitHeartBeat := func() bool {
@@ -897,7 +900,7 @@ func (ca *changeAggregator) flushBufferedEvents(ctx context.Context) error {
 // noteResolvedSpan periodically flushes Frontier progress from the current
 // changeAggregator node to the changeFrontier node to allow the changeFrontier
 // to persist the overall changefeed's progress
-func (ca *changeAggregator) noteResolvedSpan(resolved jobspb.ResolvedSpan) (returnErr error) {
+func (ca *changeAggregator) noteResolvedSpan(resolved jobspb.ResolvedSpan) error {
 	ctx, sp := tracing.ChildSpan(ca.Ctx(), "changefeed.aggregator.note_resolved_span")
 	defer sp.Finish()
 
@@ -935,32 +938,17 @@ func (ca *changeAggregator) noteResolvedSpan(resolved jobspb.ResolvedSpan) (retu
 	// TODO(yevgeniy): Consider doing something similar to how job checkpointing
 	//  works in the frontier where if we missed the window to checkpoint, we will attempt
 	//  the checkpoint at the next opportune moment.
-	checkpointFrontier := advanced &&
-		(forceFlush || timeutil.Now().After(ca.nextHighWaterFlush))
+	checkpointFrontier := (advanced && forceFlush) || ca.frontierFlushLimiter.canSave(ctx)
 
 	if checkpointFrontier {
-		defer func() {
-			ca.nextHighWaterFlush, err = nextFlushWithJitter(
-				timeutil.DefaultTimeSource{}, ca.flushFrequency, aggregatorFlushJitter.Get(sv))
-			if err != nil {
-				returnErr = errors.CombineErrors(returnErr, err)
-			}
-		}()
-		return ca.flushFrontier(ctx)
+		now := timeutil.Now()
+		if err := ca.flushFrontier(ctx); err != nil {
+			return err
+		}
+		ca.frontierFlushLimiter.doneSave(timeutil.Since(now))
 	}
 
-	// At a lower frequency, we checkpoint specific spans in the job progress
-	// either in backfills or if the highwater mark is excessively lagging behind.
-	checkpointSpans := (ca.frontier.InBackfill(resolved) || ca.frontier.HasLaggingSpans(sv)) &&
-		canCheckpointSpans(sv, ca.lastSpanFlush)
-
-	if checkpointSpans {
-		defer func() {
-			ca.lastSpanFlush = timeutil.Now()
-		}()
-		return ca.flushFrontier(ctx)
-	}
-	return returnErr
+	return nil
 }
 
 // flushFrontier flushes sink and emits resolved spans to the change frontier.
@@ -1313,8 +1301,18 @@ func newChangeFrontierProcessor(
 		cf.freqEmitResolved = emitNoResolved
 	}
 
-	cf.frontierPersistenceLimiter = newSaveRateLimiter(
-		"frontier" /* name */, changefeedbase.FrontierPersistenceInterval)
+	cf.frontierPersistenceLimiter, err = newSaveRateLimiter(saveRateConfig{
+		name: "frontier",
+		intervalName: func() redact.SafeValue {
+			return changefeedbase.FrontierPersistenceInterval.Name()
+		},
+		interval: func() time.Duration {
+			return changefeedbase.FrontierPersistenceInterval.Get(&cf.FlowCtx.Cfg.Settings.SV)
+		},
+	}, timeutil.DefaultTimeSource{})
+	if err != nil {
+		return nil, err
+	}
 
 	encodingOpts, err := opts.GetEncodingOptions()
 	if err != nil {
@@ -1919,7 +1917,7 @@ func (cf *changeFrontier) maybePersistFrontier(ctx context.Context) error {
 
 	if cf.spec.JobID == 0 ||
 		!cf.evalCtx.Settings.Version.IsActive(ctx, clusterversion.V25_4) ||
-		!cf.frontierPersistenceLimiter.canSave(ctx, &cf.FlowCtx.Cfg.Settings.SV) {
+		!cf.frontierPersistenceLimiter.canSave(ctx) {
 		return nil
 	}
 
@@ -2349,41 +2347,62 @@ func shouldCountUsageError(err error) bool {
 		status.Code(err) != codes.Canceled
 }
 
-// durationSetting is a duration cluster setting.
-type durationSetting interface {
-	Name() settings.SettingName
-	Get(sv *settings.Values) time.Duration
+// saveRateConfig is the config for a saveRateLimiter.
+type saveRateConfig struct {
+	name         redact.SafeString
+	intervalName func() redact.SafeValue
+	interval     func() time.Duration
+	jitter       func() float64 // optional
 }
 
 // saveRateLimiter is a rate limiter for saving a piece of progress.
 // It uses a duration setting as the minimum interval between saves.
 // It also limits saving to not be more frequent than the average
 // duration it takes to save progress.
 type saveRateLimiter struct {
-	name         redact.SafeString
-	saveInterval durationSetting
-	warnEveryN   util.EveryN
+	config     saveRateConfig
+	warnEveryN util.EveryN
+
+	clock timeutil.TimeSource
 
 	lastSave        time.Time
 	avgSaveDuration time.Duration
 }
 
 // newSaveRateLimiter returns a new saveRateLimiter.
-func newSaveRateLimiter(name redact.SafeString, saveInterval durationSetting) *saveRateLimiter {
-	return &saveRateLimiter{
-		name:         name,
-		saveInterval: saveInterval,
-		warnEveryN:   util.Every(time.Minute),
+func newSaveRateLimiter(
+	config saveRateConfig, clock timeutil.TimeSource,
+) (*saveRateLimiter, error) {
+	if len(config.name) == 0 {
+		return nil, errors.AssertionFailedf("name is required")
+	}
+	if config.intervalName == nil {
+		return nil, errors.AssertionFailedf("interval name is required")
+	}
+	if config.interval == nil {
+		return nil, errors.AssertionFailedf("interval is required")
 	}
+	return &saveRateLimiter{
+		config:     config,
+		warnEveryN: util.Every(time.Minute),
+		clock:      clock,
+	}, nil
 }
 
 // canSave returns whether enough time has passed to save progress again.
-func (l *saveRateLimiter) canSave(ctx context.Context, sv *settings.Values) bool {
-	interval := l.saveInterval.Get(sv)
-	if interval == 0 {
+func (l *saveRateLimiter) canSave(ctx context.Context) bool {
+	interval := l.config.interval()
+	if interval <= 0 {
 		return false
 	}
-	now := timeutil.Now()
+	if l.config.jitter != nil {
+		if jitter := l.config.jitter(); jitter > 0 {
+			if maxJitter := time.Duration(jitter * float64(interval)); maxJitter > 0 {
+				interval += time.Duration(rand.Int63n(int64(maxJitter) + 1))
+			}
+		}
+	}
+	now := l.clock.Now()
 	elapsed := now.Sub(l.lastSave)
 	if elapsed < interval {
 		return false
@@ -2393,8 +2412,7 @@ func (l *saveRateLimiter) canSave(ctx context.Context, sv *settings.Values) bool
 			log.Changefeed.Warningf(ctx, "cannot save %s even though %s has elapsed "+
 				"since last save and %s is set to %s because average duration to save was %s "+
 				"and further saving is disabled until that much time elapses",
-				l.name, elapsed, l.saveInterval.Name(),
-				interval, l.avgSaveDuration)
+				l.config.name, elapsed, l.config.intervalName(), interval, l.avgSaveDuration)
 		}
 		return false
 	}
@@ -2404,7 +2422,7 @@ func (l *saveRateLimiter) canSave(ctx context.Context, sv *settings.Values) bool
 // doneSave must be called after each save is completed with the duration
 // it took to save progress.
 func (l *saveRateLimiter) doneSave(saveDuration time.Duration) {
-	l.lastSave = timeutil.Now()
+	l.lastSave = l.clock.Now()
 
 	// Update the average save duration using an exponential moving average.
 	if l.avgSaveDuration == 0 {