Merge #142659

142659: changefeedccl: write new span-level checkpoint during ALTER CHANGEFEED r=asg0451 a=andyyang890

This patch starts writing the new span-level checkpoint during ALTER
CHANGEFEED on 25.2+ clusters.

Fixes #140509

Release note: None

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

Author: craig[bot]

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -7,13 +7,16 @@ package changefeedccl
 
 import (
 	"context"
+	"maps"
 	"net/url"
 
 	"github.com/cockroachdb/cockroach/pkg/backup/backupresolver"
 	"github.com/cockroachdb/cockroach/pkg/build"
 	"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"
@@ -154,13 +157,23 @@ func alterChangefeedPlanHook(
 		newTargets, newProgress, newStatementTime, originalSpecs, err := generateAndValidateNewTargets(
 			ctx, exprEval, p,
 			alterChangefeedStmt.Cmds,
-			newOptions.AsMap(), // TODO: Remove .AsMap()
+			newOptions,
 			prevDetails, job.Progress(),
 			newSinkURI,
 		)
 		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")
+				}
+				changefeedProgress.Checkpoint = checkpoint.ConvertToLegacyCheckpoint(changefeedProgress.SpanLevelCheckpoint)
+				changefeedProgress.SpanLevelCheckpoint = nil
+			}
+		}
 		newChangefeedStmt.Targets = newTargets
 
 		if prevDetails.Select != "" {
@@ -375,7 +388,7 @@ func generateAndValidateNewTargets(
 	exprEval exprutil.Evaluator,
 	p sql.PlanHookState,
 	alterCmds tree.AlterChangefeedCmds,
-	opts map[string]string,
+	opts changefeedbase.StatementOptions,
 	prevDetails jobspb.ChangefeedDetails,
 	prevProgress jobspb.Progress,
 	sinkURI string,
@@ -406,7 +419,11 @@ func generateAndValidateNewTargets(
 	// initial_scan set to only so that we only do an initial scan on an alter
 	// changefeed with initial_scan = 'only' if the original one also had
 	// initial_scan = 'only'.
-	_, originalInitialScanOnlyOption := opts[changefeedbase.OptInitialScanOnly]
+	originalInitialScanType, err := opts.GetInitialScanType()
+	if err != nil {
+		return nil, nil, hlc.Timestamp{}, nil, err
+	}
+	originalInitialScanOnlyOption := originalInitialScanType == changefeedbase.OnlyInitialScan
 
 	// When we add new targets with or without initial scans, indicating
 	// initial_scan or no_initial_scan in the job description would lose its
@@ -415,9 +432,9 @@ func generateAndValidateNewTargets(
 	// newly added targets, we will introduce the initial_scan opt after the
 	// job record is created.
 
-	delete(opts, changefeedbase.OptInitialScanOnly)
-	delete(opts, changefeedbase.OptNoInitialScan)
-	delete(opts, changefeedbase.OptInitialScan)
+	opts.Unset(changefeedbase.OptInitialScanOnly)
+	opts.Unset(changefeedbase.OptNoInitialScan)
+	opts.Unset(changefeedbase.OptInitialScan)
 
 	// the new progress and statement time will start from the progress and
 	// statement time of the job prior to the alteration of the changefeed. Each
@@ -445,7 +462,7 @@ func generateAndValidateNewTargets(
 
 	prevTargets := AllTargets(prevDetails)
 	noLongerExist := make(map[string]descpb.ID)
-	err = prevTargets.EachTarget(func(targetSpec changefeedbase.Target) error {
+	if err := prevTargets.EachTarget(func(targetSpec changefeedbase.Target) error {
 		k := targetKey{TableID: targetSpec.TableID, FamilyName: tree.Name(targetSpec.FamilyName)}
 		var desc catalog.TableDescriptor
 		if d, exists := descResolver.DescByID[targetSpec.TableID]; exists {
@@ -482,9 +499,7 @@ func generateAndValidateNewTargets(
 			StatementTimeName: string(targetSpec.StatementTimeName),
 		}
 		return nil
-	})
-
-	if err != nil {
+	}); err != nil {
 		return nil, nil, hlc.Timestamp{}, nil, err
 	}
 
@@ -515,25 +530,14 @@ func generateAndValidateNewTargets(
 				return nil, nil, hlc.Timestamp{}, nil, err
 			}
 
-			var withInitialScan bool
 			initialScanType, initialScanSet := targetOpts[changefeedbase.OptInitialScan]
 			_, noInitialScanSet := targetOpts[changefeedbase.OptNoInitialScan]
-			_, initialScanOnlySet := targetOpts[changefeedbase.OptInitialScanOnly]
-
-			if initialScanSet {
-				if initialScanType == `no` || (initialScanType == `only` && !originalInitialScanOnlyOption) {
-					withInitialScan = false
-				} else {
-					withInitialScan = true
-				}
-			} else {
-				withInitialScan = false
-			}
 
 			if initialScanType != `` && initialScanType != `yes` && initialScanType != `no` && initialScanType != `only` {
 				return nil, nil, hlc.Timestamp{}, nil, pgerror.Newf(
 					pgcode.InvalidParameterValue,
-					`cannot set initial_scan to %q. possible values for initial_scan are "yes", "no", "only", or no value`, changefeedbase.OptInitialScan,
+					`cannot set %q to %q. possible values for initial_scan are "yes", "no", "only", or no value`,
+					changefeedbase.OptInitialScan, initialScanType,
 				)
 			}
 
@@ -545,22 +549,11 @@ func generateAndValidateNewTargets(
 				)
 			}
 
-			if initialScanSet && initialScanOnlySet {
-				return nil, nil, hlc.Timestamp{}, nil, pgerror.Newf(
-					pgcode.InvalidParameterValue,
-					`cannot specify both %q and %q`, changefeedbase.OptInitialScan,
-					changefeedbase.OptInitialScanOnly,
-				)
-			}
-
-			if noInitialScanSet && initialScanOnlySet {
-				return nil, nil, hlc.Timestamp{}, nil, pgerror.Newf(
-					pgcode.InvalidParameterValue,
-					`cannot specify both %q and %q`, changefeedbase.OptInitialScanOnly,
-					changefeedbase.OptNoInitialScan,
-				)
-			}
+			withInitialScan := (initialScanType == `` && initialScanSet) ||
+				initialScanType == `yes` ||
+				(initialScanType == `only` && originalInitialScanOnlyOption)
 
+			// TODO(#142376): Audit whether this list is generated correctly.
 			var existingTargetIDs []descpb.ID
 			for _, targetDesc := range newTableDescs {
 				existingTargetIDs = append(existingTargetIDs, targetDesc.GetID())
@@ -659,7 +652,9 @@ func generateAndValidateNewTargets(
 			}
 		}
 		droppedTargetSpans := fetchSpansForDescs(p, droppedIDs)
-		removeSpansFromProgress(newJobProgress, droppedTargetSpans)
+		if err := removeSpansFromProgress(newJobProgress, droppedTargetSpans, newJobStatementTime); err != nil {
+			return nil, nil, hlc.Timestamp{}, nil, err
+		}
 	}
 
 	newTargetList := tree.ChangefeedTargets{}
@@ -750,7 +745,6 @@ func validateNewTargets(
 // no_initial_scan), and the current status of the job. If the progress does not
 // need to be updated, we will simply return the previous progress and statement
 // time that is passed into the function.
-// TODO(#140509): Update this function to work with the new span-level checkpoint.
 func generateNewProgress(
 	prevProgress jobspb.Progress,
 	prevStatementTime hlc.Timestamp,
@@ -760,12 +754,10 @@ func generateNewProgress(
 ) (jobspb.Progress, hlc.Timestamp, error) {
 	prevHighWater := prevProgress.GetHighWater()
 	changefeedProgress := prevProgress.GetChangefeed()
-	ptsRecord := uuid.UUID{}
-	if changefeedProgress != nil {
-		ptsRecord = changefeedProgress.ProtectedTimestampRecord
-	}
 
-	haveHighwater := !(prevHighWater == nil || prevHighWater.IsEmpty())
+	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())
 
@@ -774,6 +766,9 @@ func generateNewProgress(
 	// * the high watermark is empty, and we would like to perform an initial scan.
 	// * the high watermark is non-empty, the checkpoint is empty, and we do not want to
 	//   perform an initial scan.
+	// TODO(#142376): Consider in the scenario where we have a highwater whether
+	// we should starting sending events for the new table starting at the ALTER
+	// CHANGEFEED statement time instead of the current highwater.
 	if (!haveHighwater && withInitialScan) || (haveHighwater && !haveCheckpoint && !withInitialScan) {
 		return prevProgress, prevStatementTime, nil
 	}
@@ -788,6 +783,13 @@ func generateNewProgress(
 		)
 	}
 
+	// TODO(#142369): We should create a new PTS record instead of just
+	// copying the old one.
+	var ptsRecord uuid.UUID
+	if changefeedProgress != nil {
+		ptsRecord = changefeedProgress.ProtectedTimestampRecord
+	}
+
 	// Check if the user is trying to perform an initial scan while the high
 	// watermark is non-empty but the checkpoint is empty.
 	if haveHighwater && !haveCheckpoint && withInitialScan {
@@ -797,19 +799,18 @@ func generateNewProgress(
 		// To avoid this, we update the statement time of the job to the previous high
 		// watermark, and add all the existing targets to the checkpoint to skip the
 		// initial scan on these targets.
+		// TODO(#142376): Consider whether we want to set the new statement time
+		// to the actual new statement time (ALTER CHANGEFEED statement time).
 		newStatementTime := *prevHighWater
 
 		newProgress := jobspb.Progress{
 			Progress: &jobspb.Progress_HighWater{},
 			Details: &jobspb.Progress_Changefeed{
 				Changefeed: &jobspb.ChangefeedProgress{
-					//lint:ignore SA1019 deprecated usage
-					Checkpoint: &jobspb.ChangefeedProgress_Checkpoint{
-						Spans: existingTargetSpans,
-						// TODO(#140509): ALTER CHANGEFED should handle fine grained
-						// progress and checkpointed timestamp properly.
-					},
 					ProtectedTimestampRecord: ptsRecord,
+					SpanLevelCheckpoint: jobspb.NewTimestampSpansMap(map[hlc.Timestamp]roachpb.Spans{
+						newStatementTime: existingTargetSpans,
+					}),
 				},
 			},
 		}
@@ -826,49 +827,89 @@ func generateNewProgress(
 	// of the newly added targets so that the changefeed will skip performing
 	// a backfill on these targets.
 
-	var mergedSpanGroup roachpb.SpanGroup
-	if haveCheckpoint {
-		mergedSpanGroup.Add(changefeedProgress.Checkpoint.Spans...)
+	// TODO(#142376): In the case where we have a highwater, we will resend
+	// events since the original statement time for all existing tables.
+	// We might want to change this. We might also want to change whether
+	// the events for the new table start at the ALTER CHANGEFEED statement
+	// time instead.
+
+	spanLevelCheckpoint, err := getSpanLevelCheckpointFromProgress(prevProgress, prevStatementTime)
+	if err != nil {
+		return jobspb.Progress{}, hlc.Timestamp{}, err
 	}
-	mergedSpanGroup.Add(newSpans...)
+
+	checkpointSpansMap := maps.Collect(spanLevelCheckpoint.All())
+	var spanGroup roachpb.SpanGroup
+	spanGroup.Add(checkpointSpansMap[prevStatementTime]...)
+	spanGroup.Add(newSpans...)
+	checkpointSpansMap[prevStatementTime] = spanGroup.Slice()
 
 	newProgress := jobspb.Progress{
 		Progress: &jobspb.Progress_HighWater{},
 		Details: &jobspb.Progress_Changefeed{
 			Changefeed: &jobspb.ChangefeedProgress{
-				//lint:ignore SA1019 deprecated usage
-				Checkpoint: &jobspb.ChangefeedProgress_Checkpoint{
-					Spans: mergedSpanGroup.Slice(),
-				},
 				ProtectedTimestampRecord: ptsRecord,
+				SpanLevelCheckpoint:      jobspb.NewTimestampSpansMap(checkpointSpansMap),
 			},
 		},
 	}
 	return newProgress, prevStatementTime, nil
 }
 
-// TODO(#140509): Update this function to work with the new span-level checkpoint.
-func removeSpansFromProgress(prevProgress jobspb.Progress, spansToRemove []roachpb.Span) {
+func removeSpansFromProgress(
+	prevProgress jobspb.Progress, spansToRemove []roachpb.Span, statementTime hlc.Timestamp,
+) error {
 	changefeedProgress := prevProgress.GetChangefeed()
 	if changefeedProgress == nil {
-		return
+		return nil
 	}
 	changefeedCheckpoint := changefeedProgress.Checkpoint
 	if changefeedCheckpoint == nil {
-		return
+		return nil
 	}
-	prevSpans := changefeedCheckpoint.Spans
 
-	var spanGroup roachpb.SpanGroup
-	spanGroup.Add(prevSpans...)
-	spanGroup.Sub(spansToRemove...)
-	changefeedProgress.Checkpoint.Spans = spanGroup.Slice()
+	spanLevelCheckpoint, err := getSpanLevelCheckpointFromProgress(prevProgress, statementTime)
+	if err != nil {
+		return err
+	}
+	checkpointSpansMap := make(map[hlc.Timestamp]roachpb.Spans)
+	for ts, sp := range spanLevelCheckpoint.All() {
+		var spanGroup roachpb.SpanGroup
+		spanGroup.Add(sp...)
+		spanGroup.Sub(spansToRemove...)
+		if spans := spanGroup.Slice(); len(spans) > 0 {
+			checkpointSpansMap[ts] = spans
+		}
+	}
+	changefeedProgress.SpanLevelCheckpoint = jobspb.NewTimestampSpansMap(checkpointSpansMap)
+
+	return nil
+}
+
+func getSpanLevelCheckpointFromProgress(
+	progress jobspb.Progress, statementTime hlc.Timestamp,
+) (*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
 }
 
-func fetchSpansForDescs(p sql.PlanHookState, droppedIDs []descpb.ID) (primarySpans []roachpb.Span) {
+func fetchSpansForDescs(p sql.PlanHookState, descIDs []descpb.ID) (primarySpans []roachpb.Span) {
 	seen := make(map[descpb.ID]struct{})
 	codec := p.ExtendedEvalContext().Codec
-	for _, id := range droppedIDs {
+	for _, id := range descIDs {
 		if _, isDup := seen[id]; isDup {
 			continue
 		}