Merge #153491

153491: changefeedccl: periodically write entire span frontier to job_info r=KeithCh a=andyyang890

**changefeedccl: periodically write entire span frontier to job_info** 

Fixes #153200

Release note (performance improvement): Changefeeds will now
periodically persist their entire span frontiers so that
fewer duplicates will need to be emitted during restarts.
The default persistence interval is 30s, but this can be
configured with the `changefeed.progress.frontier_persistence.interval`
cluster setting.

---

**roachtest: add benchmark for changefeed frontier persistence** 

Fixes #153202

Release note: None

Co-authored-by: Andy Yang <[email protected]>

Author: craig[bot]

docs/generated/metrics/metrics.yaml

@@ -1837,6 +1837,14 @@ layers:
       unit: NANOSECONDS
       aggregation: AVG
       derivative: NONE
+    - name: changefeed.stage.frontier_persistence.latency
+      exported_name: changefeed_stage_frontier_persistence_latency
+      description: 'Latency of the changefeed stage: persisting frontier to job info'
+      y_axis_label: Latency
+      type: HISTOGRAM
+      unit: NANOSECONDS
+      aggregation: AVG
+      derivative: NONE
     - name: changefeed.stage.kv_feed_buffer.latency
       exported_name: changefeed_stage_kv_feed_buffer_latency
       description: 'Latency of the changefeed stage: waiting to buffer kv events'

docs/generated/settings/settings-for-tenants.txt

@@ -22,6 +22,7 @@ changefeed.kafka_v2_error_details.enabled	boolean	true	if enabled, Kafka v2 sink
 changefeed.memory.per_changefeed_limit	byte size	512 MiB	controls amount of data that can be buffered per changefeed	application
 changefeed.resolved_timestamp.min_update_interval (alias: changefeed.min_highwater_advance)	duration	0s	minimum amount of time that must have elapsed since the last time a changefeed's resolved timestamp was updated before it is eligible to be updated again; default of 0 means no minimum interval is enforced but updating will still be limited by the average time it takes to checkpoint progress	application
 changefeed.node_throttle_config	string		specifies node level throttling configuration for all changefeeeds	application
+changefeed.progress.frontier_persistence.interval	duration	30s	minimum amount of time that must elapse before a changefeed will persist its entire span frontier again	application
 changefeed.protect_timestamp.max_age	duration	96h0m0s	fail the changefeed if the protected timestamp age exceeds this threshold; 0 disables expiration	application
 changefeed.protect_timestamp_interval	duration	10m0s	controls how often the changefeed forwards its protected timestamp to the resolved timestamp	application
 changefeed.schema_feed.read_with_priority_after	duration	1m0s	retry with high priority if we were not able to read descriptors for too long; 0 disables	application

docs/generated/settings/settings.html

@@ -27,6 +27,7 @@
 <tr><td><div id="setting-changefeed-memory-per-changefeed-limit" class="anchored"><code>changefeed.memory.per_changefeed_limit</code></div></td><td>byte size</td><td><code>512 MiB</code></td><td>controls amount of data that can be buffered per changefeed</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-min-highwater-advance" class="anchored"><code>changefeed.resolved_timestamp.min_update_interval<br />(alias: changefeed.min_highwater_advance)</code></div></td><td>duration</td><td><code>0s</code></td><td>minimum amount of time that must have elapsed since the last time a changefeed&#39;s resolved timestamp was updated before it is eligible to be updated again; default of 0 means no minimum interval is enforced but updating will still be limited by the average time it takes to checkpoint progress</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-node-throttle-config" class="anchored"><code>changefeed.node_throttle_config</code></div></td><td>string</td><td><code></code></td><td>specifies node level throttling configuration for all changefeeeds</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
+<tr><td><div id="setting-changefeed-progress-frontier-persistence-interval" class="anchored"><code>changefeed.progress.frontier_persistence.interval</code></div></td><td>duration</td><td><code>30s</code></td><td>minimum amount of time that must elapse before a changefeed will persist its entire span frontier again</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-protect-timestamp-max-age" class="anchored"><code>changefeed.protect_timestamp.max_age</code></div></td><td>duration</td><td><code>96h0m0s</code></td><td>fail the changefeed if the protected timestamp age exceeds this threshold; 0 disables expiration</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-protect-timestamp-interval" class="anchored"><code>changefeed.protect_timestamp_interval</code></div></td><td>duration</td><td><code>10m0s</code></td><td>controls how often the changefeed forwards its protected timestamp to the resolved timestamp</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-schema-feed-read-with-priority-after" class="anchored"><code>changefeed.schema_feed.read_with_priority_after</code></div></td><td>duration</td><td><code>1m0s</code></td><td>retry with high priority if we were not able to read descriptors for too long; 0 disables</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>

pkg/ccl/changefeedccl/BUILD.bazel

@@ -74,6 +74,7 @@ go_library(
         "//pkg/docs",
         "//pkg/featureflag",
         "//pkg/jobs",
+        "//pkg/jobs/jobfrontier",
         "//pkg/jobs/jobsauth",
         "//pkg/jobs/jobspb",
         "//pkg/jobs/jobsprofiler",
@@ -127,6 +128,7 @@ go_library(
         "//pkg/sql/sessiondatapb",
         "//pkg/sql/syntheticprivilege",
         "//pkg/sql/types",
+        "//pkg/util",
         "//pkg/util/admission",
         "//pkg/util/admission/admissionpb",
         "//pkg/util/bufalloc",
@@ -201,6 +203,7 @@ go_test(
         "alter_changefeed_test.go",
         "changefeed_dist_test.go",
         "changefeed_job_info_test.go",
+        "changefeed_progress_test.go",
         "changefeed_stmt_test.go",
         "changefeed_test.go",
         "csv_test.go",
@@ -262,6 +265,7 @@ go_test(
         "//pkg/geo/geopb",
         "//pkg/internal/sqlsmith",
         "//pkg/jobs",
+        "//pkg/jobs/jobfrontier",
         "//pkg/jobs/jobspb",
         "//pkg/jobs/jobstest",
         "//pkg/keys",

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -24,6 +24,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/schemafeed"
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobfrontier"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/kvcoord"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts"
@@ -41,6 +42,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/util"
 	bulkutil "github.com/cockroachdb/cockroach/pkg/util/bulk"
 	"github.com/cockroachdb/cockroach/pkg/util/cancelchecker"
 	"github.com/cockroachdb/cockroach/pkg/util/envutil"
@@ -1055,6 +1057,9 @@ type changeFrontier struct {
 	// record was updated to the frontier's highwater mark
 	lastProtectedTimestampUpdate time.Time
 
+	// frontierPersistenceLimiter is a rate limiter for persisting the span frontier.
+	frontierPersistenceLimiter *saveRateLimiter
+
 	// js, if non-nil, is called to checkpoint the changefeed's
 	// progress in the corresponding system job entry.
 	js *jobState
@@ -1302,6 +1307,9 @@ func newChangeFrontierProcessor(
 		cf.freqEmitResolved = emitNoResolved
 	}
 
+	cf.frontierPersistenceLimiter = newSaveRateLimiter(
+		"frontier" /* name */, changefeedbase.FrontierPersistenceInterval)
+
 	encodingOpts, err := opts.GetEncodingOptions()
 	if err != nil {
 		return nil, err
@@ -1803,33 +1811,33 @@ func (cf *changeFrontier) maybeCheckpointJob(
 			return false, nil
 		}
 		checkpointStart := timeutil.Now()
-		updated, err := cf.checkpointJobProgress(ctx, cf.frontier.Frontier(), checkpoint, cf.evalCtx.Settings.Version)
-		if err != nil {
+		if err := cf.checkpointJobProgress(ctx, cf.frontier.Frontier(), checkpoint); err != nil {
 			return false, err
 		}
 		cf.js.checkpointCompleted(ctx, timeutil.Since(checkpointStart))
-		return updated, nil
 	}
 
-	return false, nil
+	if err := cf.maybePersistFrontier(ctx); err != nil {
+		return false, err
+	}
+
+	// TODO(#153462): Determine if this return value should return true
+	// only if the highwater was updated.
+	return updateCheckpoint || updateHighWater, nil
 }
 
 const changefeedJobProgressTxnName = "changefeed job progress"
 
 func (cf *changeFrontier) checkpointJobProgress(
-	ctx context.Context,
-	frontier hlc.Timestamp,
-	spanLevelCheckpoint *jobspb.TimestampSpansMap,
-	cv clusterversion.Handle,
-) (bool, error) {
+	ctx context.Context, frontier hlc.Timestamp, spanLevelCheckpoint *jobspb.TimestampSpansMap,
+) error {
 	ctx, sp := tracing.ChildSpan(ctx, "changefeed.frontier.checkpoint_job_progress")
 	defer sp.Finish()
 	defer cf.sliMetrics.Timers.CheckpointJobProgress.Start()()
 
 	if cf.knobs.RaiseRetryableError != nil {
 		if err := cf.knobs.RaiseRetryableError(); err != nil {
-			return false, changefeedbase.MarkRetryableError(
-				errors.New("cf.knobs.RaiseRetryableError"))
+			return changefeedbase.MarkRetryableError(errors.New("cf.knobs.RaiseRetryableError"))
 		}
 	}
 
@@ -1857,6 +1865,9 @@ func (cf *changeFrontier) checkpointJobProgress(
 			changefeedProgress := progress.Details.(*jobspb.Progress_Changefeed).Changefeed
 			changefeedProgress.SpanLevelCheckpoint = spanLevelCheckpoint
 
+			// TODO(#153299): Make sure we only updated per-table PTS if we persisted
+			// the span frontier. We'll probably want to move this code out of
+			// checkpointJobProgress and into maybeCheckpointJob.
 			if ptsUpdated, err = cf.manageProtectedTimestamps(ctx, txn, changefeedProgress); err != nil {
 				log.Changefeed.Warningf(ctx, "error managing protected timestamp record: %v", err)
 				return err
@@ -1884,7 +1895,7 @@ func (cf *changeFrontier) checkpointJobProgress(
 
 			return nil
 		}); err != nil {
-			return false, err
+			return err
 		}
 		if ptsUpdated {
 			cf.lastProtectedTimestampUpdate = timeutil.Now()
@@ -1898,7 +1909,28 @@ func (cf *changeFrontier) checkpointJobProgress(
 	cf.localState.SetHighwater(frontier)
 	cf.localState.SetCheckpoint(spanLevelCheckpoint)
 
-	return true, nil
+	return nil
+}
+
+func (cf *changeFrontier) maybePersistFrontier(ctx context.Context) error {
+	ctx, sp := tracing.ChildSpan(ctx, "changefeed.frontier.maybe_persist_frontier")
+	defer sp.Finish()
+
+	if cf.spec.JobID == 0 ||
+		!cf.evalCtx.Settings.Version.IsActive(ctx, clusterversion.V25_4) ||
+		!cf.frontierPersistenceLimiter.canSave(ctx, &cf.FlowCtx.Cfg.Settings.SV) {
+		return nil
+	}
+
+	timer := cf.sliMetrics.Timers.FrontierPersistence.Start()
+	if err := cf.FlowCtx.Cfg.DB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+		return jobfrontier.Store(ctx, txn, cf.spec.JobID, "coordinator", cf.frontier)
+	}); err != nil {
+		return err
+	}
+	persistDuration := timer()
+	cf.frontierPersistenceLimiter.doneSave(persistDuration)
+	return nil
 }
 
 // manageProtectedTimestamps periodically advances the protected timestamp for
@@ -2315,3 +2347,70 @@ func shouldCountUsageError(err error) bool {
 		pgerror.GetPGCode(err) != pgcode.UndefinedTable &&
 		status.Code(err) != codes.Canceled
 }
+
+// durationSetting is a duration cluster setting.
+type durationSetting interface {
+	Name() settings.SettingName
+	Get(sv *settings.Values) time.Duration
+}
+
+// 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
+
+	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),
+	}
+}
+
+// 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 {
+		return false
+	}
+	now := timeutil.Now()
+	elapsed := now.Sub(l.lastSave)
+	if elapsed < interval {
+		return false
+	}
+	if elapsed < l.avgSaveDuration {
+		if l.warnEveryN.ShouldProcess(now) {
+			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)
+		}
+		return false
+	}
+	return true
+}
+
+// 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()
+
+	// Update the average save duration using an exponential moving average.
+	if l.avgSaveDuration == 0 {
+		l.avgSaveDuration = saveDuration
+	} else {
+		const alpha = 0.1
+		l.avgSaveDuration = time.Duration(
+			alpha*float64(saveDuration) + (1-alpha)*float64(l.avgSaveDuration))
+	}
+}

pkg/ccl/changefeedccl/changefeed_progress_test.go

@@ -0,0 +1,93 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package changefeedccl
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdctest"
+	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobfrontier"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/sql/isql"
+	"github.com/cockroachdb/cockroach/pkg/testutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+// TestChangefeedFrontierPersistence verifies that changefeeds periodically
+// persist their span frontiers to the job info table.
+func TestChangefeedFrontierPersistence(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+		ctx := context.Background()
+
+		// Set a short interval for frontier persistence.
+		sqlDB.Exec(t, "SET CLUSTER SETTING changefeed.progress.frontier_persistence.interval = '5s'")
+
+		// Get frontier persistence metric.
+		registry := s.Server.JobRegistry().(*jobs.Registry)
+		metric := registry.MetricsStruct().Changefeed.(*Metrics).AggMetrics.Timers.FrontierPersistence
+
+		// Verify metric count starts at zero.
+		initialCount, _ := metric.CumulativeSnapshot().Total()
+		require.Equal(t, int64(0), initialCount)
+
+		// Create a table and insert some data.
+		sqlDB.Exec(t, "CREATE TABLE foo (a INT PRIMARY KEY, b STRING)")
+		sqlDB.Exec(t, "INSERT INTO foo VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+
+		// Start a changefeed.
+		foo := feed(t, f, "CREATE CHANGEFEED FOR foo")
+		defer closeFeed(t, foo)
+
+		// Make sure frontier gets persisted to job_info table.
+		jobID := foo.(cdctest.EnterpriseTestFeed).JobID()
+		testutils.SucceedsSoon(t, func() error {
+			var found bool
+			var allSpans []jobspb.ResolvedSpan
+			if err := s.Server.InternalDB().(isql.DB).Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+				var err error
+				allSpans, found, err = jobfrontier.GetAllResolvedSpans(ctx, txn, jobID)
+				if err != nil {
+					return err
+				}
+				return nil
+			}); err != nil {
+				return err
+			}
+			if !found {
+				return errors.Newf("frontier not yet persisted")
+			}
+			t.Logf("found resolved spans in job_info table: %+v", allSpans)
+			return nil
+		})
+
+		// Verify metric count and average latency have sensible values.
+		testutils.SucceedsSoon(t, func() error {
+			metricSnapshot := metric.CumulativeSnapshot()
+			count, _ := metricSnapshot.Total()
+			if count == 0 {
+				return errors.Newf("metrics not yet updated")
+			}
+			avgLatency := time.Duration(metricSnapshot.Mean())
+			t.Logf("frontier persistence metrics - count: %d, avg latency: %s", count, avgLatency)
+			require.Greater(t, count, int64(0))
+			require.Greater(t, avgLatency, time.Duration(0))
+			return nil
+		})
+	}
+
+	cdcTest(t, testFn, feedTestEnterpriseSinks)
+}

pkg/ccl/changefeedccl/changefeedbase/settings.go

@@ -420,3 +420,15 @@ var TrackPerTableProgress = settings.RegisterBoolSetting(
 		"but doing so may incur additional overhead during ordinary changefeed execution",
 	metamorphic.ConstantWithTestBool("changefeed.progress.per_table_tracking.enabled", true),
 )
+
+// FrontierPersistenceInterval configures the minimum amount of time that must
+// elapse before a changefeed will persist its entire span frontier again.
+var FrontierPersistenceInterval = settings.RegisterDurationSettingWithExplicitUnit(
+	settings.ApplicationLevel,
+	"changefeed.progress.frontier_persistence.interval",
+	"minimum amount of time that must elapse before a changefeed "+
+		"will persist its entire span frontier again",
+	30*time.Second, /* defaultValue */
+	settings.DurationInRange(5*time.Second, 10*time.Minute),
+	settings.WithPublic,
+)