Merge #148757

148757: changefeedccl: delete legacy checkpoint code r=asg0451,rharding6373 a=andyyang890

Since 25.2 is a required upgrade, we can now safely delete all of the
legacy checkpoint code.

Fixes #139734

Release note: None

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

Author: craig[bot]

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -15,8 +15,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdceval"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedvalidators"
-	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/checkpoint"
-	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobsauth"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
@@ -161,18 +159,6 @@ func alterChangefeedPlanHook(
 		if err != nil {
 			return err
 		}
-		if !p.ExecCfg().Settings.Version.IsActive(ctx, clusterversion.V25_2) {
-			changefeedProgress := newProgress.GetChangefeed()
-			if changefeedProgress != nil && !changefeedProgress.SpanLevelCheckpoint.IsEmpty() {
-				if !changefeedProgress.Checkpoint.IsEmpty() {
-					return errors.AssertionFailedf("both legacy and current checkpoint set on changefeed job progress")
-				}
-				legacyCheckpoint := checkpoint.ConvertToLegacyCheckpoint(changefeedProgress.SpanLevelCheckpoint)
-				if err := changefeedProgress.SetCheckpoint(legacyCheckpoint, nil); err != nil {
-					return err
-				}
-			}
-		}
 		newChangefeedStmt.TableTargets = newTargets
 
 		if prevDetails.Select != "" {
@@ -653,7 +639,7 @@ func generateAndValidateNewTargets(
 			}
 		}
 		droppedTargetSpans := fetchSpansForDescs(p, droppedIDs)
-		if err := removeSpansFromProgress(newJobProgress, droppedTargetSpans, newJobStatementTime); err != nil {
+		if err := removeSpansFromProgress(newJobProgress, droppedTargetSpans); err != nil {
 			return nil, nil, hlc.Timestamp{}, nil, err
 		}
 	}
@@ -759,8 +745,7 @@ func generateNewProgress(
 	haveHighwater := prevHighWater != nil && prevHighWater.IsSet()
 	// TODO(#142376): Whether a checkpoint exists seems orthogonal to what
 	// we do in this function. Consider removing this flag.
-	haveCheckpoint := changefeedProgress != nil &&
-		(!changefeedProgress.Checkpoint.IsEmpty() || !changefeedProgress.SpanLevelCheckpoint.IsEmpty())
+	haveCheckpoint := changefeedProgress != nil && !changefeedProgress.SpanLevelCheckpoint.IsEmpty()
 
 	// Check if the progress does not need to be updated. The progress does not
 	// need to be updated if:
@@ -834,7 +819,7 @@ func generateNewProgress(
 	// the events for the new table start at the ALTER CHANGEFEED statement
 	// time instead.
 
-	spanLevelCheckpoint, err := getSpanLevelCheckpointFromProgress(prevProgress, prevStatementTime)
+	spanLevelCheckpoint, err := getSpanLevelCheckpointFromProgress(prevProgress)
 	if err != nil {
 		return jobspb.Progress{}, hlc.Timestamp{}, err
 	}
@@ -857,10 +842,8 @@ func generateNewProgress(
 	return newProgress, prevStatementTime, nil
 }
 
-func removeSpansFromProgress(
-	progress jobspb.Progress, spansToRemove []roachpb.Span, statementTime hlc.Timestamp,
-) error {
-	spanLevelCheckpoint, err := getSpanLevelCheckpointFromProgress(progress, statementTime)
+func removeSpansFromProgress(progress jobspb.Progress, spansToRemove []roachpb.Span) error {
+	spanLevelCheckpoint, err := getSpanLevelCheckpointFromProgress(progress)
 	if err != nil {
 		return err
 	}
@@ -876,31 +859,18 @@ func removeSpansFromProgress(
 			checkpointSpansMap[ts] = spans
 		}
 	}
-	spanLevelCheckpoint = jobspb.NewTimestampSpansMap(checkpointSpansMap)
-	if err := progress.GetChangefeed().SetCheckpoint(nil, spanLevelCheckpoint); err != nil {
-		return err
-	}
+	progress.GetChangefeed().SpanLevelCheckpoint = jobspb.NewTimestampSpansMap(checkpointSpansMap)
 
 	return nil
 }
 
 func getSpanLevelCheckpointFromProgress(
-	progress jobspb.Progress, statementTime hlc.Timestamp,
+	progress jobspb.Progress,
 ) (*jobspb.TimestampSpansMap, error) {
 	changefeedProgress := progress.GetChangefeed()
 	if changefeedProgress == nil {
 		return nil, nil
 	}
-
-	if !changefeedProgress.Checkpoint.IsEmpty() && !changefeedProgress.SpanLevelCheckpoint.IsEmpty() {
-		return nil, errors.AssertionFailedf("both legacy and current checkpoint set on changefeed job progress")
-	}
-
-	if legacyCheckpoint := changefeedProgress.Checkpoint; !legacyCheckpoint.IsEmpty() {
-		hw := progress.GetHighWater()
-		return checkpoint.ConvertFromLegacyCheckpoint(legacyCheckpoint, statementTime, *hw), nil
-	}
-
 	return changefeedProgress.SpanLevelCheckpoint, nil
 }
 

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -13,9 +13,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdceval"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
-	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/checkpoint"
 	"github.com/cockroachdb/cockroach/pkg/ccl/kvccl/kvfollowerreadsccl"
-	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobsprofiler"
 	"github.com/cockroachdb/cockroach/pkg/keys"
@@ -260,38 +258,15 @@ func startDistChangefeed(
 
 	dsp := execCtx.DistSQLPlanner()
 
-	//lint:ignore SA1019 deprecated usage
-	var legacyCheckpoint *jobspb.ChangefeedProgress_Checkpoint
-	if progress := localState.progress.GetChangefeed(); progress != nil && progress.Checkpoint != nil {
-		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 always 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.
-		if buildutil.CrdbTestBuild {
-			return errors.AssertionFailedf("both legacy and current checkpoint set on " +
-				"changefeed job progress")
+		if log.V(2) {
+			log.Infof(ctx, "span-level checkpoint: %s", spanLevelCheckpoint)
 		}
-		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, legacyCheckpoint, spanLevelCheckpoint, localState.drainingNodes)(ctx, dsp)
+		trackedSpans, spanLevelCheckpoint, localState.drainingNodes)(ctx, dsp)
 	if err != nil {
 		return err
 	}
@@ -410,8 +385,6 @@ func makePlan(
 	description string,
 	initialHighWater hlc.Timestamp,
 	trackedSpans []roachpb.Span,
-	//lint:ignore SA1019 deprecated usage
-	legacyCheckpoint *jobspb.ChangefeedProgress_Checkpoint,
 	spanLevelCheckpoint *jobspb.TimestampSpansMap,
 	drainingNodes []roachpb.NodeID,
 ) func(context.Context, *sql.DistSQLPlanner) (*sql.PhysicalPlan, *sql.PlanningCtx, error) {
@@ -480,62 +453,22 @@ func makePlan(
 			maybeCfKnobs.SpanPartitionsCallback(spanPartitions)
 		}
 
-		// Use the same checkpoint for all aggregators; each aggregator will only look at
-		// spans that are assigned to it.
-		// We could compute per-aggregator checkpoint, but that's probably an overkill.
-		//lint:ignore SA1019 deprecated usage
-		var aggregatorCheckpoint execinfrapb.ChangeAggregatorSpec_Checkpoint
-		var checkpointSpanGroup roachpb.SpanGroup
-
-		if legacyCheckpoint != nil {
-			checkpointSpanGroup.Add(legacyCheckpoint.Spans...)
-			aggregatorCheckpoint.Spans = legacyCheckpoint.Spans
-			aggregatorCheckpoint.Timestamp = legacyCheckpoint.Timestamp
-		}
-		if log.V(2) {
-			log.Infof(ctx, "aggregator checkpoint: %s", aggregatorCheckpoint)
-		}
-
 		aggregatorSpecs := make([]*execinfrapb.ChangeAggregatorSpec, len(spanPartitions))
 		for i, sp := range spanPartitions {
 			if log.ExpensiveLogEnabled(ctx, 2) {
 				log.Infof(ctx, "watched spans for node %d: %v", sp.SQLInstanceID, sp)
 			}
-			watches := make([]execinfrapb.ChangeAggregatorSpec_Watch, len(sp.Spans))
 
-			var initialHighWaterPtr *hlc.Timestamp
+			watches := make([]execinfrapb.ChangeAggregatorSpec_Watch, len(sp.Spans))
 			for watchIdx, nodeSpan := range sp.Spans {
-				if evalCtx.Settings.Version.IsActive(ctx, clusterversion.V25_2) {
-					// If the cluster has been fully upgraded to v25.2, we should populate
-					// the initial highwater of ChangeAggregatorSpec_Watch and leave the
-					// initial resolved of each span empty. We rely on the aggregators to
-					// forward the checkpointed timestamp for every span based on
-					// aggregatorCheckpoint.
-					watches[watchIdx] = execinfrapb.ChangeAggregatorSpec_Watch{
-						Span: nodeSpan,
-					}
-					initialHighWaterPtr = &initialHighWater
-				} else {
-					// If the cluster has not been fully upgraded to v25.2, we should
-					// leave the initial highwater of ChangeAggregatorSpec_Watch as nil.
-					// We rely on this to tell the aggregators to the initial resolved
-					// timestamp for each span to infer the initial highwater. Read more
-					// from changeAggregator.getInitialHighWaterAndSpans.
-					initialResolved := initialHighWater
-					if checkpointSpanGroup.Encloses(nodeSpan) {
-						initialResolved = legacyCheckpoint.Timestamp
-					}
-					watches[watchIdx] = execinfrapb.ChangeAggregatorSpec_Watch{
-						Span:            nodeSpan,
-						InitialResolved: initialResolved,
-					}
+				watches[watchIdx] = execinfrapb.ChangeAggregatorSpec_Watch{
+					Span: nodeSpan,
 				}
 			}
 
 			aggregatorSpecs[i] = &execinfrapb.ChangeAggregatorSpec{
 				Watches:             watches,
-				Checkpoint:          aggregatorCheckpoint,
-				InitialHighWater:    initialHighWaterPtr,
+				InitialHighWater:    &initialHighWater,
 				SpanLevelCheckpoint: spanLevelCheckpoint,
 				Feed:                details,
 				UserProto:           execCtx.User().EncodeProto(),
@@ -550,17 +483,12 @@ func makePlan(
 		// is created, even if it is paused and unpaused, but #28982 describes some
 		// ways that this might happen in the future.
 		changeFrontierSpec := execinfrapb.ChangeFrontierSpec{
-			TrackedSpans: trackedSpans,
-			Feed:         details,
-			JobID:        jobID,
-			UserProto:    execCtx.User().EncodeProto(),
-			Description:  description,
-		}
-
-		if spanLevelCheckpoint != nil {
-			changeFrontierSpec.SpanLevelCheckpoint = spanLevelCheckpoint
-		} else {
-			changeFrontierSpec.SpanLevelCheckpoint = checkpoint.ConvertFromLegacyCheckpoint(legacyCheckpoint, details.StatementTime, initialHighWater)
+			TrackedSpans:        trackedSpans,
+			SpanLevelCheckpoint: spanLevelCheckpoint,
+			Feed:                details,
+			JobID:               jobID,
+			UserProto:           execCtx.User().EncodeProto(),
+			Description:         description,
 		}
 
 		if haveKnobs && maybeCfKnobs.OnDistflowSpec != nil {