changefeedccl: protect system tables with their own pts record

Previously, we were relying on each pts record that protects
a user table to also protect system tables. This would mean
that we'd protect the system tables once for each table.

Now, we have a dedicated pts record for system tables that
protects back to the highwater and do not include the system
table protections in the table specific records. We only
protect the system tables once.

Fixes: #152446
Epic: CRDB-1421

Release note: None

Author: aerfrei

pkg/ccl/changefeedccl/BUILD.bazel

@@ -281,6 +281,7 @@ go_test(
         "//pkg/kv/kvserver/kvserverbase",
         "//pkg/kv/kvserver/protectedts",
         "//pkg/kv/kvserver/protectedts/ptpb",
+        "//pkg/kv/kvserver/protectedts/ptutil",
         "//pkg/multitenant/tenantcapabilitiespb",
         "//pkg/roachpb",
         "//pkg/scheduledjobs",

pkg/ccl/changefeedccl/cdcprogresspb/progress.proto

@@ -9,11 +9,20 @@ option go_package = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcp
 
 import "gogoproto/gogo.proto";
 
-// ProtectedTimestampRecords is a map from table descriptor IDs to protected timestamp record IDs.
+// ProtectedTimestampRecords contains the IDs of protected timestamps records
+// for a changefeed.
 message ProtectedTimestampRecords {
-  map<uint32, bytes> protected_timestamp_records = 1 [
+  // UserTables is a map from the ID of a user table to the protected timestamp
+  // record that protects it.
+  map<uint32, bytes> user_tables = 1 [
     (gogoproto.castkey) = "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb.ID",
     (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/uuid.UUID",
     (gogoproto.nullable) = false
   ];
+  // SystemTables is the ID of PTS record that protects the system tables
+  // back to the changefeed's highwater.
+  bytes system_tables = 2 [
+    (gogoproto.nullable) = false,
+    (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/uuid.UUID"
+  ];
 }

pkg/ccl/changefeedccl/changefeed_job_info_test.go

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

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -1999,7 +1999,19 @@ func (cf *changeFrontier) manageProtectedTimestamps(
 	}()
 
 	if cf.spec.ProgressConfig != nil && cf.spec.ProgressConfig.PerTableProtectedTimestamps {
-		return cf.managePerTableProtectedTimestamps(ctx, txn, &ptsEntries, highwater)
+		updatedPerTablePTS, err :=
+			cf.managePerTableProtectedTimestamps(ctx, txn, &ptsEntries, highwater, pts)
+		if err != nil {
+			return false, err
+		}
+
+		updatedSystemTablesPTS, err :=
+			cf.advanceSystemTablesProtectedTimestamp(ctx, txn, &ptsEntries, highwater, pts)
+		if err != nil {
+			return false, err
+		}
+
+		return updatedPerTablePTS || updatedSystemTablesPTS, nil
 	}
 
 	return cf.advanceProtectedTimestamp(ctx, progress, pts, highwater)
@@ -2010,8 +2022,8 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 	txn isql.Txn,
 	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
 	highwater hlc.Timestamp,
+	pts protectedts.Storage,
 ) (updatedPerTablePTS bool, err error) {
-	pts := cf.FlowCtx.Cfg.ProtectedTimestampProvider.WithTxn(txn)
 	tableIDsToCreate := make(map[descpb.ID]hlc.Timestamp)
 	for tableID, frontier := range cf.frontier.Frontiers() {
 		tableHighWater := func() hlc.Timestamp {
@@ -2023,14 +2035,13 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 			return frontier.Frontier()
 		}()
 
-		if ptsEntries.ProtectedTimestampRecords[tableID] != uuid.Nil {
+		if ptsEntries.UserTables[tableID] != uuid.Nil {
 			if updated, err := cf.advancePerTableProtectedTimestampRecord(ctx, ptsEntries, tableID, tableHighWater, pts); err != nil {
 				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
@@ -2061,7 +2072,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.UserTables[tableID])
 	if err != nil {
 		return false, err
 	}
@@ -2071,7 +2082,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.UserTables[tableID], tableHighWater); err != nil {
 		return false, err
 	}
 	return true, nil
@@ -2083,19 +2094,19 @@ func (cf *changeFrontier) createPerTableProtectedTimestampRecords(
 	tableIDsToCreate map[descpb.ID]hlc.Timestamp,
 	pts protectedts.Storage,
 ) error {
-	if ptsEntries.ProtectedTimestampRecords == nil {
-		ptsEntries.ProtectedTimestampRecords = make(map[descpb.ID]uuid.UUID)
+	if ptsEntries.UserTables == nil {
+		ptsEntries.UserTables = make(map[descpb.ID]uuid.UUID)
 	}
 	for tableID, tableHighWater := range tableIDsToCreate {
 		targets, err := cf.createPerTablePTSTargets(tableID)
 		if err != nil {
 			return err
 		}
-		ptr := createProtectedTimestampRecord(
+		ptr := createUserTablesProtectedTimestampRecord(
 			ctx, cf.FlowCtx.Codec(), cf.spec.JobID, targets, tableHighWater,
 		)
 		uuid := ptr.ID.GetUUID()
-		ptsEntries.ProtectedTimestampRecords[tableID] = uuid
+		ptsEntries.UserTables[tableID] = uuid
 		if err := pts.Protect(ctx, ptr); err != nil {
 			return err
 		}
@@ -2124,14 +2135,63 @@ func (cf *changeFrontier) createPerTablePTSTargets(
 	return targets, nil
 }
 
+func (cf *changeFrontier) advanceSystemTablesProtectedTimestamp(
+	ctx context.Context,
+	txn isql.Txn,
+	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
+	timestamp hlc.Timestamp,
+	pts protectedts.Storage,
+) (updated bool, err error) {
+	if ptsEntries.SystemTables == uuid.Nil {
+		// All changefeeds using per-table PTS records should have a system tables
+		// PTS record. If they are missing one, it should be made when starting the
+		// changefeed.
+		return false, errors.AssertionFailedf("expected system tables PTS record to be present")
+	}
+
+	rec, err := pts.GetRecord(ctx, ptsEntries.SystemTables)
+	if err != nil {
+		return false, err
+	}
+
+	if !makeSystemTablesTargetToProtect().Equal(rec.Target) {
+		if cf.knobs.PreservePTSTargets != nil && cf.knobs.PreservePTSTargets() {
+			return false, nil
+		}
+		if err := cf.remakeSystemTablesPTSRecord(ctx, txn, pts, ptsEntries, timestamp); err != nil {
+			return false, err
+		}
+		log.VEventf(
+			ctx, 2, "remade system tables PTS record %v to include all targets",
+			ptsEntries.SystemTables,
+		)
+		return true, nil
+	}
+
+	ptsUpdateLag := changefeedbase.ProtectTimestampLag.Get(&cf.FlowCtx.Cfg.Settings.SV)
+	if rec.Timestamp.AddDuration(ptsUpdateLag).After(timestamp) {
+		return false, nil
+	}
+
+	if err := pts.UpdateTimestamp(ctx, ptsEntries.SystemTables, timestamp); err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+// advanceProtectedTimestamp advances the single PTS record for changefeeds that
+// are not using per-table protected timestamps.
 func (cf *changeFrontier) advanceProtectedTimestamp(
 	ctx context.Context,
 	progress *jobspb.ChangefeedProgress,
 	pts protectedts.Storage,
 	timestamp hlc.Timestamp,
 ) (updated bool, err error) {
 	if progress.ProtectedTimestampRecord == uuid.Nil {
-		ptr := createProtectedTimestampRecord(
+		// For changefeeds not using per-table PTS, system tables are protected
+		// in the single PTS record for the changefeed with all other targets
+		// in a combined record.
+		ptr := createCombinedProtectedTimestampRecord(
 			ctx, cf.FlowCtx.Codec(), cf.spec.JobID, cf.targets, timestamp,
 		)
 		progress.ProtectedTimestampRecord = ptr.ID.GetUUID()
@@ -2159,7 +2219,7 @@ func (cf *changeFrontier) advanceProtectedTimestamp(
 	// If we've identified more tables that need to be protected since this
 	// changefeed was created, it will be missing here. If so, we "migrate" it
 	// to include all the appropriate targets.
-	if !makeTargetToProtect(cf.targets).Equal(rec.Target) {
+	if !makeCombinedTargetToProtect(cf.targets).Equal(rec.Target) {
 		if preservePTSTargets := cf.knobs.PreservePTSTargets != nil && cf.knobs.PreservePTSTargets(); preservePTSTargets {
 			return false, nil
 		}
@@ -2190,7 +2250,7 @@ func (cf *changeFrontier) remakePTSRecord(
 	resolved hlc.Timestamp,
 ) error {
 	prevRecordId := progress.ProtectedTimestampRecord
-	ptr := createProtectedTimestampRecord(
+	ptr := createCombinedProtectedTimestampRecord(
 		ctx, cf.FlowCtx.Codec(), cf.spec.JobID, cf.targets, resolved,
 	)
 	if err := pts.Protect(ctx, ptr); err != nil {
@@ -2207,6 +2267,29 @@ func (cf *changeFrontier) remakePTSRecord(
 	return nil
 }
 
+func (cf *changeFrontier) remakeSystemTablesPTSRecord(
+	ctx context.Context,
+	txn isql.Txn,
+	pts protectedts.Storage,
+	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
+	resolved hlc.Timestamp,
+) error {
+	ptr := createSystemTablesProtectedTimestampRecord(
+		ctx, cf.FlowCtx.Codec(), cf.spec.JobID, resolved,
+	)
+	if err := pts.Protect(ctx, ptr); err != nil {
+		return err
+	}
+	prevRecordId := ptsEntries.SystemTables
+	if err := pts.Release(ctx, prevRecordId); err != nil {
+		return err
+	}
+	ptsEntries.SystemTables = ptr.ID.GetUUID()
+	log.Eventf(ctx, "created new system tables pts record %v to replace old pts record %v at %v",
+		ptsEntries.SystemTables, prevRecordId, resolved)
+	return writeChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, ptsEntries, txn, cf.spec.JobID)
+}
+
 func (cf *changeFrontier) maybeEmitResolved(ctx context.Context, newResolved hlc.Timestamp) error {
 	if cf.freqEmitResolved == emitNoResolved || newResolved.IsEmpty() {
 		return nil

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -327,6 +327,9 @@ func changefeedPlanHook(
 			// perTablePTSRecords is the per-table protected timestamp records which exist when
 			// per-table protected timestamps are enabled.
 			var perTablePTSRecords []*ptpb.Record
+			// systemTablesPTSRecord is the system tables protected timestamp record which exists when
+			// per-table protected timestamps are enabled.
+			var systemTablesPTSRecord *ptpb.Record
 			// ptsRecords is the protected timestamp records object containing all per-table protected
 			// timestamp records. Its format matches what will be persisted to the job info table.
 			var ptsRecords *cdcprogresspb.ProtectedTimestampRecords
@@ -339,9 +342,11 @@ func changefeedPlanHook(
 			if usingPerTablePTS {
 				protectedTimestampRecords := make(map[descpb.ID]uuid.UUID)
 				if err := targets.EachTarget(func(target changefeedbase.Target) error {
+					// TODO(#155957): We are likely leaking PTS records here in
+					// the column families case.
 					ptsTargets := changefeedbase.Targets{}
 					ptsTargets.Add(target)
-					ptsRecord := createProtectedTimestampRecord(
+					ptsRecord := createUserTablesProtectedTimestampRecord(
 						ctx,
 						codec,
 						jobID,
@@ -355,11 +360,18 @@ func changefeedPlanHook(
 				}); err != nil {
 					return err
 				}
+				systemTablesPTSRecord = createSystemTablesProtectedTimestampRecord(
+					ctx,
+					codec,
+					jobID,
+					details.StatementTime,
+				)
 				ptsRecords = &cdcprogresspb.ProtectedTimestampRecords{
-					ProtectedTimestampRecords: protectedTimestampRecords,
+					UserTables:   protectedTimestampRecords,
+					SystemTables: systemTablesPTSRecord.ID.GetUUID(),
 				}
 			} else {
-				ptr = createProtectedTimestampRecord(
+				ptr = createCombinedProtectedTimestampRecord(
 					ctx,
 					codec,
 					jobID,
@@ -380,12 +392,15 @@ func changefeedPlanHook(
 					}
 				}
 				if usingPerTablePTS {
+					pts := p.ExecCfg().ProtectedTimestampProvider.WithTxn(p.InternalSQLTxn())
 					for _, perTableRecord := range perTablePTSRecords {
-						pts := p.ExecCfg().ProtectedTimestampProvider.WithTxn(p.InternalSQLTxn())
 						if err := pts.Protect(ctx, perTableRecord); err != nil {
 							return err
 						}
 					}
+					if err := pts.Protect(ctx, systemTablesPTSRecord); err != nil {
+						return err
+					}
 					if err := writeChangefeedJobInfo(
 						ctx, perTableProtectedTimestampsFilename, ptsRecords, p.InternalSQLTxn(), jobID,
 					); err != nil {
@@ -428,6 +443,9 @@ func changefeedPlanHook(
 							return err
 						}
 					}
+					if err := pts.Protect(ctx, systemTablesPTSRecord); err != nil {
+						return err
+					}
 					if err := writeChangefeedJobInfo(
 						ctx, perTableProtectedTimestampsFilename, ptsRecords, txn, jobID,
 					); err != nil {
@@ -1966,20 +1984,27 @@ func (b *changefeedResumer) OnFailOrCancel(
 	}
 
 	maybeCleanUpProtectedTimestamp(progress.GetChangefeed().ProtectedTimestampRecord)
+	// We clean up the per-table protected timestamps (and their accompanying
+	// system tables protected timestamp record) in a transaction since we need
+	// to read from the job info.
 	if err := execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 		ptsEntries := cdcprogresspb.ProtectedTimestampRecords{}
 		if err := readChangefeedJobInfo(
 			ctx, perTableProtectedTimestampsFilename, &ptsEntries, txn, b.job.ID(),
 		); err != nil {
 			return err
 		}
-
-		if len(ptsEntries.ProtectedTimestampRecords) == 0 {
+		// In the event that the changefeed is not using per-table protected
+		// timestamps, the ptsEntries populated from the job info table
+		// (in the file perTableProtectedTimestampsFilename) will be empty.
+		// There is nothing to clean up, so we can safely return here.
+		if len(ptsEntries.UserTables) == 0 && ptsEntries.SystemTables == uuid.Nil {
 			return nil
 		}
-		for _, record := range ptsEntries.ProtectedTimestampRecords {
+		for _, record := range ptsEntries.UserTables {
 			maybeCleanUpProtectedTimestamp(record)
 		}
+		maybeCleanUpProtectedTimestamp(ptsEntries.SystemTables)
 		return deleteChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, txn, b.job.ID())
 	}); err != nil {
 		return err

pkg/ccl/changefeedccl/changefeed_test.go

@@ -12103,6 +12103,9 @@ func TestChangefeedAvroDecimalColumnWithDiff(t *testing.T) {
 	cdcTest(t, testFn, feedTestForceSink("kafka"))
 }
 
+// TestChangefeedProtectedTimestampUpdate tests that a changefeed will update
+// its protected timestamp record when the high watermark advances and update
+// the related metrics.
 func TestChangefeedProtectedTimestampUpdate(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -12177,7 +12180,7 @@ WITH resolved='10ms', min_checkpoint_frequency='10ms', no_initial_scan`
 				); err != nil {
 					return err
 				}
-				ptsRecordID := ptsEntries.ProtectedTimestampRecords[descpb.ID(tableID)]
+				ptsRecordID := ptsEntries.UserTables[descpb.ID(tableID)]
 				ts = getTimestampFromPTSRecord(ptsRecordID)
 				return nil
 			})