changefeedccl: remove shared PTS record for non-lagging tables

Previously, per-table PTS changefeeds used a single protected
timestamp record for all non-lagging tables, while assigning
individual records only to lagging ones. This commit removes
that shared record; instead, all tables now receive their own
per-table PTS record at startup, each updated using the table's
highwater.

Epic: CRDB-1421
Release note: None

Author: aerfrei

pkg/ccl/changefeedccl/cdcprogresspb/progress.proto

@@ -13,6 +13,7 @@ import "gogoproto/gogo.proto";
 message ProtectedTimestampRecords {
   map<uint32, bytes> protected_timestamp_records = 1 [
     (gogoproto.castkey) = "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb.ID",
-    (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/uuid.UUID"
+    (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/uuid.UUID",
+    (gogoproto.nullable) = false
   ];
 }

pkg/ccl/changefeedccl/changefeed_job_info.go

@@ -45,6 +45,20 @@ func writeChangefeedJobInfo(
 	return jobs.WriteChunkedFileToJobInfo(ctx, filename, data, txn, jobID)
 }
 
+// deleteChangefeedJobInfo deletes a changefeed job info protobuf from the
+// job_info table. A changefeed-specific filename is required.
+func deleteChangefeedJobInfo(
+	ctx context.Context, filename string, txn isql.Txn, jobID jobspb.JobID,
+) error {
+	if !strings.HasPrefix(filename, jobInfoFilenamePrefix) {
+		return errors.AssertionFailedf("filename %q must start with prefix %q", filename, jobInfoFilenamePrefix)
+	}
+	if !strings.HasSuffix(filename, jobInfoFilenameExtension) {
+		return errors.AssertionFailedf("filename %q must end with extension %q", filename, jobInfoFilenameExtension)
+	}
+	return jobs.WriteChunkedFileToJobInfo(ctx, filename, nil, txn, jobID)
+}
+
 // readChangefeedJobInfo reads a changefeed job info protobuf from the
 // job_info table. A changefeed-specific filename is required.
 func readChangefeedJobInfo(

pkg/ccl/changefeedccl/changefeed_job_info_test.go

@@ -35,9 +35,9 @@ func TestChangefeedJobInfoRoundTrip(t *testing.T) {
 	uuid1 := uuid.MakeV4()
 	uuid2 := uuid.MakeV4()
 	ptsRecords := cdcprogresspb.ProtectedTimestampRecords{
-		ProtectedTimestampRecords: map[descpb.ID]*uuid.UUID{
-			descpb.ID(100): &uuid1,
-			descpb.ID(200): &uuid2,
+		ProtectedTimestampRecords: map[descpb.ID]uuid.UUID{
+			descpb.ID(100): uuid1,
+			descpb.ID(200): uuid2,
 		},
 	}
 

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -1957,6 +1957,10 @@ func (cf *changeFrontier) manageProtectedTimestamps(
 		}
 	}()
 
+	if cf.knobs.ManagePTSError != nil {
+		return false, cf.knobs.ManagePTSError()
+	}
+
 	var ptsEntries cdcprogresspb.ProtectedTimestampRecords
 	if err := readChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, &ptsEntries, txn, cf.spec.JobID); err != nil {
 		return false, err
@@ -1971,15 +1975,7 @@ func (cf *changeFrontier) manageProtectedTimestamps(
 	}()
 
 	if cf.spec.ProgressConfig.PerTableProtectedTimestamps {
-		newPTS, updatedPerTablePTS, err := cf.managePerTableProtectedTimestamps(ctx, txn, &ptsEntries, highwater)
-		if err != nil {
-			return false, err
-		}
-		updatedMainPTS, err := cf.advanceProtectedTimestamp(ctx, progress, pts, newPTS)
-		if err != nil {
-			return false, err
-		}
-		return updatedMainPTS || updatedPerTablePTS, nil
+		return cf.managePerTableProtectedTimestamps(ctx, txn, &ptsEntries, highwater)
 	}
 
 	return cf.advanceProtectedTimestamp(ctx, progress, pts, highwater)
@@ -1990,28 +1986,8 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 	txn isql.Txn,
 	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
 	highwater hlc.Timestamp,
-) (newPTS hlc.Timestamp, updatedPerTablePTS bool, err error) {
-	var leastLaggingTimestamp hlc.Timestamp
-	for _, frontier := range cf.frontier.Frontiers() {
-		if frontier.Frontier().After(leastLaggingTimestamp) {
-			leastLaggingTimestamp = frontier.Frontier()
-		}
-	}
-
-	newPTS = func() hlc.Timestamp {
-		lagDuration := changefeedbase.ProtectTimestampBucketingInterval.Get(&cf.FlowCtx.Cfg.Settings.SV)
-		ptsLagCutoff := leastLaggingTimestamp.AddDuration(-lagDuration)
-		// If we are within the bucketing interval of having started the changefeed,
-		// we use the highwater as the PTS timestamp so as not to try to protect
-		// tables before the changefeed started.
-		if ptsLagCutoff.Less(highwater) {
-			return highwater
-		}
-		return ptsLagCutoff
-	}()
-
+) (updatedPerTablePTS bool, err error) {
 	pts := cf.FlowCtx.Cfg.ProtectedTimestampProvider.WithTxn(txn)
-	tableIDsToRelease := make([]descpb.ID, 0)
 	tableIDsToCreate := make(map[descpb.ID]hlc.Timestamp)
 	for tableID, frontier := range cf.frontier.Frontiers() {
 		tableHighWater := func() hlc.Timestamp {
@@ -2023,66 +1999,35 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 			return frontier.Frontier()
 		}()
 
-		isLagging := tableHighWater.Less(newPTS)
-
-		if cf.knobs.IsTableLagging != nil && cf.knobs.IsTableLagging(tableID) {
-			isLagging = true
-		}
-
-		if !isLagging {
-			if ptsEntries.ProtectedTimestampRecords[tableID] != nil {
-				tableIDsToRelease = append(tableIDsToRelease, tableID)
-			}
-			continue
-		}
-
-		if ptsEntries.ProtectedTimestampRecords[tableID] != nil {
+		if ptsEntries.ProtectedTimestampRecords[tableID] != uuid.Nil {
 			if updated, err := cf.advancePerTableProtectedTimestampRecord(ctx, ptsEntries, tableID, tableHighWater, pts); err != nil {
-				return hlc.Timestamp{}, false, err
+				return false, err
 			} else if updated {
 				updatedPerTablePTS = true
 			}
 		} else {
 			// TODO(#152448): Do not include system table protections in these records.
+			// TODO(#153894): Newly added/dropped tables should be caught and
+			// protected when starting the frontier, not here.
 			tableIDsToCreate[tableID] = tableHighWater
 		}
 	}
 
-	if len(tableIDsToRelease) > 0 {
-		if err := cf.releasePerTableProtectedTimestampRecords(ctx, ptsEntries, tableIDsToRelease, pts); err != nil {
-			return hlc.Timestamp{}, false, err
-		}
-	}
-
 	if len(tableIDsToCreate) > 0 {
-		if err := cf.createPerTableProtectedTimestampRecords(ctx, ptsEntries, tableIDsToCreate, pts); err != nil {
-			return hlc.Timestamp{}, false, err
+		if err := cf.createPerTableProtectedTimestampRecords(
+			ctx, ptsEntries, tableIDsToCreate, pts,
+		); err != nil {
+			return false, err
 		}
-	}
-
-	if len(tableIDsToRelease) > 0 || len(tableIDsToCreate) > 0 {
-		if err := writeChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, ptsEntries, txn, cf.spec.JobID); err != nil {
-			return hlc.Timestamp{}, false, err
+		if err := writeChangefeedJobInfo(
+			ctx, perTableProtectedTimestampsFilename, ptsEntries, txn, cf.spec.JobID,
+		); err != nil {
+			return false, err
 		}
 		updatedPerTablePTS = true
 	}
 
-	return newPTS, updatedPerTablePTS, nil
-}
-
-func (cf *changeFrontier) releasePerTableProtectedTimestampRecords(
-	ctx context.Context,
-	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
-	tableIDs []descpb.ID,
-	pts protectedts.Storage,
-) error {
-	for _, tableID := range tableIDs {
-		if err := pts.Release(ctx, *ptsEntries.ProtectedTimestampRecords[tableID]); err != nil {
-			return err
-		}
-		delete(ptsEntries.ProtectedTimestampRecords, tableID)
-	}
-	return nil
+	return updatedPerTablePTS, nil
 }
 
 func (cf *changeFrontier) advancePerTableProtectedTimestampRecord(
@@ -2092,7 +2037,7 @@ func (cf *changeFrontier) advancePerTableProtectedTimestampRecord(
 	tableHighWater hlc.Timestamp,
 	pts protectedts.Storage,
 ) (updated bool, err error) {
-	rec, err := pts.GetRecord(ctx, *ptsEntries.ProtectedTimestampRecords[tableID])
+	rec, err := pts.GetRecord(ctx, ptsEntries.ProtectedTimestampRecords[tableID])
 	if err != nil {
 		return false, err
 	}
@@ -2102,7 +2047,7 @@ func (cf *changeFrontier) advancePerTableProtectedTimestampRecord(
 		return false, nil
 	}
 
-	if err := pts.UpdateTimestamp(ctx, *ptsEntries.ProtectedTimestampRecords[tableID], tableHighWater); err != nil {
+	if err := pts.UpdateTimestamp(ctx, ptsEntries.ProtectedTimestampRecords[tableID], tableHighWater); err != nil {
 		return false, err
 	}
 	return true, nil
@@ -2115,7 +2060,7 @@ func (cf *changeFrontier) createPerTableProtectedTimestampRecords(
 	pts protectedts.Storage,
 ) error {
 	if ptsEntries.ProtectedTimestampRecords == nil {
-		ptsEntries.ProtectedTimestampRecords = make(map[descpb.ID]*uuid.UUID)
+		ptsEntries.ProtectedTimestampRecords = make(map[descpb.ID]uuid.UUID)
 	}
 	for tableID, tableHighWater := range tableIDsToCreate {
 		targets, err := cf.createPerTablePTSTargets(tableID)
@@ -2126,7 +2071,7 @@ func (cf *changeFrontier) createPerTableProtectedTimestampRecords(
 			ctx, cf.FlowCtx.Codec(), cf.spec.JobID, targets, tableHighWater,
 		)
 		uuid := ptr.ID.GetUUID()
-		ptsEntries.ProtectedTimestampRecords[tableID] = &uuid
+		ptsEntries.ProtectedTimestampRecords[tableID] = uuid
 		if err := pts.Protect(ctx, ptr); err != nil {
 			return err
 		}
@@ -2210,10 +2155,6 @@ func (cf *changeFrontier) advanceProtectedTimestamp(
 		return false, nil
 	}
 
-	if cf.knobs.ManagePTSError != nil {
-		return false, cf.knobs.ManagePTSError()
-	}
-
 	log.VEventf(ctx, 2, "updating protected timestamp %v at %v", progress.ProtectedTimestampRecord, timestamp)
 	return true, pts.UpdateTimestamp(ctx, progress.ProtectedTimestampRecord, timestamp)
 }