Merge #153421 #154135

153421: changefeedccl: remove shared PTS record for non-lagging tables r=log-head,andyyang890 a=aerfrei

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.

Fixes #153681

Epic: [CRDB-1421](https://cockroachlabs.atlassian.net/browse/CRDB-1421)
Release note: None

154135: sql/inspect: skip INSPECT checks for non-applicable spans earlier r=spilchen a=spilchen

INSPECT jobs generate a set of checks along with a separate list of spans, which only apply to some of the checks. Previously, filtering to ensure a span applied to a check happened late in the process.

This commit moves that filtering earlier, so checks immediately skip spans that are not relevant. This improves efficiency and prepares the system for upcoming progress tracking logic, which will be added in a separate PR.

Informs: #148297
Epic: CRDB-30356

Release note: none

Co-authored-by: Aerin Freilich <[email protected]>
Co-authored-by: Matt Spilchen <[email protected]>

Author: 3 people

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)
 }