changefeedccl: fix job progress checkpoint fields invariant violation

Release note (bug fix): A bug where a changefeed that was created before
v25.2 could fail after upgrading to v25.2 with the error message
`both legacy and current checkpoint set on change aggregator spec`
has now been fixed.

Author: andyyang890

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -170,8 +170,10 @@ func alterChangefeedPlanHook(
 				if !changefeedProgress.Checkpoint.IsEmpty() {
 					return errors.AssertionFailedf("both legacy and current checkpoint set on changefeed job progress")
 				}
-				changefeedProgress.Checkpoint = checkpoint.ConvertToLegacyCheckpoint(changefeedProgress.SpanLevelCheckpoint)
-				changefeedProgress.SpanLevelCheckpoint = nil
+				legacyCheckpoint := checkpoint.ConvertToLegacyCheckpoint(changefeedProgress.SpanLevelCheckpoint)
+				if err := changefeedProgress.SetCheckpoint(legacyCheckpoint, nil); err != nil {
+					return err
+				}
 			}
 		}
 		newChangefeedStmt.Targets = newTargets
@@ -859,21 +861,15 @@ func generateNewProgress(
 }
 
 func removeSpansFromProgress(
-	prevProgress jobspb.Progress, spansToRemove []roachpb.Span, statementTime hlc.Timestamp,
+	progress jobspb.Progress, spansToRemove []roachpb.Span, statementTime hlc.Timestamp,
 ) error {
-	changefeedProgress := prevProgress.GetChangefeed()
-	if changefeedProgress == nil {
-		return nil
-	}
-	changefeedCheckpoint := changefeedProgress.Checkpoint
-	if changefeedCheckpoint == nil {
-		return nil
-	}
-
-	spanLevelCheckpoint, err := getSpanLevelCheckpointFromProgress(prevProgress, statementTime)
+	spanLevelCheckpoint, err := getSpanLevelCheckpointFromProgress(progress, statementTime)
 	if err != nil {
 		return err
 	}
+	if spanLevelCheckpoint == nil {
+		return nil
+	}
 	checkpointSpansMap := make(map[hlc.Timestamp]roachpb.Spans)
 	for ts, sp := range spanLevelCheckpoint.All() {
 		var spanGroup roachpb.SpanGroup
@@ -883,7 +879,10 @@ func removeSpansFromProgress(
 			checkpointSpansMap[ts] = spans
 		}
 	}
-	changefeedProgress.SpanLevelCheckpoint = jobspb.NewTimestampSpansMap(checkpointSpansMap)
+	spanLevelCheckpoint = jobspb.NewTimestampSpansMap(checkpointSpansMap)
+	if err := progress.GetChangefeed().SetCheckpoint(nil, spanLevelCheckpoint); err != nil {
+		return err
+	}
 
 	return nil
 }

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -261,16 +261,33 @@ func startDistChangefeed(
 	dsp := execCtx.DistSQLPlanner()
 
 	//lint:ignore SA1019 deprecated usage
-	var checkpoint *jobspb.ChangefeedProgress_Checkpoint
+	var legacyCheckpoint *jobspb.ChangefeedProgress_Checkpoint
 	if progress := localState.progress.GetChangefeed(); progress != nil && progress.Checkpoint != nil {
-		checkpoint = progress.Checkpoint
+		legacyCheckpoint = progress.Checkpoint
 	}
 	var spanLevelCheckpoint *jobspb.TimestampSpansMap
 	if progress := localState.progress.GetChangefeed(); progress != nil && progress.SpanLevelCheckpoint != nil {
 		spanLevelCheckpoint = progress.SpanLevelCheckpoint
 	}
+	if legacyCheckpoint != nil && spanLevelCheckpoint != nil {
+		if legacyCheckpoint.Timestamp.After(spanLevelCheckpoint.MinTimestamp()) {
+			// We should never be writing the legacy checkpoint again once we
+			// start writing the new checkpoint format. If we do, that signals
+			// a missing or incorrect version gate check somewhere.
+			return errors.AssertionFailedf("both legacy and current checkpoint set on " +
+				"changefeed job progress and legacy checkpoint has later timestamp")
+		}
+		// This should be an assertion failure but unfortunately due to a bug
+		// that was included in earlier versions of 25.2 (#148620), we may fail
+		// to clear the legacy checkpoint when we start writing the new one.
+		// We instead discard the legacy checkpoint here and it will eventually be
+		// cleared once the cluster is running a newer patch release with the fix.
+		log.Warningf(ctx, "both legacy and current checkpoint set on changefeed job progress; "+
+			"discarding legacy checkpoint")
+		legacyCheckpoint = nil
+	}
 	p, planCtx, err := makePlan(execCtx, jobID, details, description, initialHighWater,
-		trackedSpans, checkpoint, spanLevelCheckpoint, localState.drainingNodes)(ctx, dsp)
+		trackedSpans, legacyCheckpoint, spanLevelCheckpoint, localState.drainingNodes)(ctx, dsp)
 	if err != nil {
 		return err
 	}

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -1148,11 +1148,11 @@ func (cs *cachedState) SetHighwater(frontier hlc.Timestamp) {
 }
 
 // SetCheckpoint implements the eval.ChangefeedState interface.
-func (cs *cachedState) SetCheckpoint(checkpoint *jobspb.TimestampSpansMap) {
-	// NB: It's not necessary to set the legacy checkpoint field because this
+func (cs *cachedState) SetCheckpoint(checkpoint *jobspb.TimestampSpansMap) error {
+	// It's not necessary to do a version gate check here because this
 	// copy of the checkpoint is only used in-memory on a coordinator node that
 	// knows about the new field.
-	cs.progress.Details.(*jobspb.Progress_Changefeed).Changefeed.SpanLevelCheckpoint = checkpoint
+	return cs.progress.Details.(*jobspb.Progress_Changefeed).Changefeed.SetCheckpoint(nil, checkpoint)
 }
 
 // AggregatorFrontierSpans returns an iterator over the spans in the aggregator
@@ -1864,10 +1864,14 @@ func (cf *changeFrontier) checkpointJobProgress(
 
 			changefeedProgress := progress.Details.(*jobspb.Progress_Changefeed).Changefeed
 			if cv.IsActive(cf.Ctx(), clusterversion.V25_2) {
-				changefeedProgress.SpanLevelCheckpoint = spanLevelCheckpoint
+				if err := changefeedProgress.SetCheckpoint(nil, spanLevelCheckpoint); err != nil {
+					return err
+				}
 			} else {
 				legacyCheckpoint = checkpoint.ConvertToLegacyCheckpoint(spanLevelCheckpoint)
-				changefeedProgress.Checkpoint = legacyCheckpoint
+				if err := changefeedProgress.SetCheckpoint(legacyCheckpoint, nil); err != nil {
+					return err
+				}
 			}
 
 			if ptsUpdated, err = cf.manageProtectedTimestamps(ctx, txn, changefeedProgress); err != nil {
@@ -1900,7 +1904,9 @@ func (cf *changeFrontier) checkpointJobProgress(
 	}
 
 	cf.localState.SetHighwater(frontier)
-	cf.localState.SetCheckpoint(spanLevelCheckpoint)
+	if err := cf.localState.SetCheckpoint(spanLevelCheckpoint); err != nil {
+		return false, err
+	}
 
 	return true, nil
 }

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -1642,7 +1642,9 @@ func reconcileJobStateWithLocalState(
 		if updateHW {
 			localState.SetHighwater(sf.Frontier())
 		}
-		localState.SetCheckpoint(checkpoint)
+		if err := localState.SetCheckpoint(checkpoint); err != nil {
+			return err
+		}
 		if log.V(1) {
 			log.Infof(ctx, "Applying checkpoint to job record:  hw=%v, cf=%v",
 				localState.progress.GetHighWater(), localState.progress.GetChangefeed())

pkg/ccl/changefeedccl/changefeed_test.go

@@ -11703,3 +11703,83 @@ func TestCloudstorageParallelCompression(t *testing.T) {
 		}
 	})
 }
+
+// TestChangefeedResumeWithBothLegacyAndCurrentCheckpoint is a regression
+// test for #148620, which was a bug where the legacy checkpoint was not
+// being cleared after the cluster was upgraded to 25.2 and subsequently
+// causing an assertion error when resuming.
+func TestChangefeedResumeWithBothLegacyAndCurrentCheckpoint(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		ctx := context.Background()
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+
+		// Create a table with 11 ranges.
+		const numRows = 10
+		const numRanges = numRows + 1
+		sqlDB.ExecMultiple(t,
+			`CREATE TABLE foo (a INT PRIMARY KEY)`,
+			fmt.Sprintf(`INSERT INTO foo SELECT * FROM generate_series(1, %d)`, numRows),
+			fmt.Sprintf(`ALTER TABLE foo SPLIT AT (SELECT * FROM generate_series(1, %d))`, numRows),
+		)
+		fooSpan := desctestutils.
+			TestingGetPublicTableDescriptor(s.Server.DB(), s.Codec, "d", "foo").
+			PrimaryIndexSpan(s.Codec)
+		ranges, _, err := s.Server.
+			DistSenderI().(*kvcoord.DistSender).
+			AllRangeSpans(ctx, []roachpb.Span{fooSpan})
+		require.NoError(t, err)
+		require.Len(t, ranges, numRanges)
+
+		cf := feed(t, f, `CREATE CHANGEFEED FOR foo WITH no_initial_scan`)
+		defer closeFeed(t, cf)
+
+		jobFeed, ok := cf.(cdctest.EnterpriseTestFeed)
+		require.True(t, ok)
+
+		sqlDB.Exec(t, `PAUSE JOB $1`, jobFeed.JobID())
+		waitForJobState(sqlDB, t, jobFeed.JobID(), jobs.StatePaused)
+		hw, err := jobFeed.HighWaterMark()
+		require.NoError(t, err)
+
+		registry := s.Server.JobRegistry().(*jobs.Registry)
+
+		// Manually insert both a legacy and current checkpoint containing
+		// a random range from the table.
+		rnd, _ := randutil.NewTestRand()
+		randomRange := ranges[rnd.Intn(len(ranges))]
+		err = registry.UpdateJobWithTxn(ctx, jobFeed.JobID(), nil,
+			func(txn isql.Txn, md jobs.JobMetadata, ju *jobs.JobUpdater) error {
+				checkpointTS := hw.Add(int64(time.Nanosecond), 0)
+				progress := md.Progress
+				progress.Details = jobspb.WrapProgressDetails(jobspb.ChangefeedProgress{
+					//lint:ignore SA1019 deprecated usage
+					Checkpoint: &jobspb.ChangefeedProgress_Checkpoint{
+						Spans:     []roachpb.Span{randomRange},
+						Timestamp: checkpointTS,
+					},
+					SpanLevelCheckpoint: jobspb.NewTimestampSpansMap(
+						map[hlc.Timestamp]roachpb.Spans{
+							checkpointTS: []roachpb.Span{randomRange},
+						},
+					),
+				})
+				ju.UpdateProgress(progress)
+				return nil
+			})
+		require.NoError(t, err)
+
+		sqlDB.Exec(t, `RESUME JOB $1`, jobFeed.JobID())
+		waitForJobState(sqlDB, t, jobFeed.JobID(), jobs.StateRunning)
+
+		// Wait for highwater to advance past the current time.
+		var tsStr string
+		sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&tsStr)
+		ts := parseTimeToHLC(t, tsStr)
+		require.NoError(t, jobFeed.WaitForHighWaterMark(ts))
+	}
+
+	cdcTest(t, testFn, feedTestEnterpriseSinks)
+}

pkg/jobs/jobspb/jobs.go

@@ -16,6 +16,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/iterutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing/tracingpb"
+	"github.com/cockroachdb/errors"
 )
 
 // JobID is the ID of a job.
@@ -152,6 +153,20 @@ func (m *ChangefeedProgress_Checkpoint) IsEmpty() bool {
 	return m == nil || (len(m.Spans) == 0 && m.Timestamp.IsEmpty())
 }
 
+// SetCheckpoint is a setter for the checkpoint fields in ChangefeedProgress.
+// It enforces the invariant that at most one of them can be non-nil.
+func (m *ChangefeedProgress) SetCheckpoint(
+	legacyCheckpoint *ChangefeedProgress_Checkpoint, spanLevelCheckpoint *TimestampSpansMap,
+) error {
+	if legacyCheckpoint != nil && spanLevelCheckpoint != nil {
+		return errors.AssertionFailedf(
+			"attempting to set both legacy and current checkpoint on changefeed job progress")
+	}
+	m.Checkpoint = legacyCheckpoint
+	m.SpanLevelCheckpoint = spanLevelCheckpoint
+	return nil
+}
+
 func (r RestoreDetails) OnlineImpl() bool {
 	return r.ExperimentalCopy || r.ExperimentalOnline
 }

pkg/jobs/jobspb/jobs.proto

@@ -1217,6 +1217,10 @@ message ChangefeedProgress {
   // changefeed. It consists of a list of spans and a single timestamp
   // representing the minimum resolved timestamp of the spans.
   // It is now deprecated in favor of SpanLevelCheckpoint.
+  //
+  // Invariant: At most one of Checkpoint and SpanLevelCheckpoint should be
+  // non-nil at any time.
+  //
   // TODO(#139734): Delete this field and mark field number as reserved.
   Checkpoint checkpoint = 4 [deprecated=true];
 
@@ -1243,6 +1247,9 @@ message ChangefeedProgress {
   // checkpoint to its corresponding resolved timestamp, which will be higher
   // than the overall resolved timestamp and thus allow us to do less work.
   // This is especially useful during backfills or if some spans are lagging.
+  //
+  // Invariant: At most one of Checkpoint and SpanLevelCheckpoint should be
+  // non-nil at any time.
   TimestampSpansMap span_level_checkpoint = 5;
 }
 

pkg/sql/sem/eval/deps.go

@@ -665,7 +665,7 @@ type ChangefeedState interface {
 	SetHighwater(frontier hlc.Timestamp)
 
 	// SetCheckpoint sets the checkpoint for the changefeed.
-	SetCheckpoint(checkpoint *jobspb.TimestampSpansMap)
+	SetCheckpoint(checkpoint *jobspb.TimestampSpansMap) error
 }
 
 // TenantOperator is capable of interacting with tenant state, allowing SQL