changefeedccl: improve target creation for lagging tables

Previously, we would iterate through all feed targets to
create the pts targets for a lagging tables, which was
very inefficient. We improve this efficiency by using the
underlying map by tableID. We also do only a single write
to the jobInfo table instead of one for each record we
create.

Informs: #147788
Epic: CRDB-1421

Release note: None

Author: aerfrei

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -1985,6 +1985,7 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 
 	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 {
 			// If this table has not yet finished its initial scan, we use the highwater
@@ -2016,10 +2017,7 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 			}
 		} else {
 			// TODO(#152448): Do not include system table protections in these records.
-			if err := cf.createPerTableProtectedTimestampRecord(ctx, txn, ptsEntries, tableID, tableHighWater, pts); err != nil {
-				return hlc.Timestamp{}, false, err
-			}
-			updatedPerTablePTS = true
+			tableIDsToCreate[tableID] = tableHighWater
 		}
 	}
 
@@ -2030,6 +2028,13 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 		updatedPerTablePTS = true
 	}
 
+	if len(tableIDsToCreate) > 0 {
+		if err := cf.createPerTableProtectedTimestampRecords(ctx, txn, ptsEntries, tableIDsToCreate, pts); err != nil {
+			return hlc.Timestamp{}, false, err
+		}
+		updatedPerTablePTS = true
+	}
+
 	return newPTS, updatedPerTablePTS, nil
 }
 
@@ -2072,40 +2077,51 @@ func (cf *changeFrontier) advancePerTableProtectedTimestampRecord(
 	return true, nil
 }
 
-func (cf *changeFrontier) createPerTableProtectedTimestampRecord(
+func (cf *changeFrontier) createPerTableProtectedTimestampRecords(
 	ctx context.Context,
 	txn isql.Txn,
 	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
-	tableID descpb.ID,
-	tableHighWater hlc.Timestamp,
+	tableIDsToCreate map[descpb.ID]hlc.Timestamp,
 	pts protectedts.Storage,
 ) error {
-	// If the table is lagging and doesn't have a per table PTS record,
-	// we create a new one.
-	targets := changefeedbase.Targets{}
-	if cf.targets.Size > 0 {
-		err := cf.targets.EachTarget(func(target changefeedbase.Target) error {
-			if target.DescID == tableID {
-				targets.Add(target)
-			}
-			return nil
-		})
+	if ptsEntries.ProtectedTimestampRecords == nil {
+		ptsEntries.ProtectedTimestampRecords = make(map[descpb.ID]*uuid.UUID)
+	}
+	for tableID, tableHighWater := range tableIDsToCreate {
+		targets, err := cf.createPerTablePTSTarget(tableID)
 		if err != nil {
 			return err
 		}
+		ptr := createProtectedTimestampRecord(
+			ctx, cf.FlowCtx.Codec(), cf.spec.JobID, targets, tableHighWater,
+		)
+		uuid := ptr.ID.GetUUID()
+		ptsEntries.ProtectedTimestampRecords[tableID] = &uuid
+		if err := pts.Protect(ctx, ptr); err != nil {
+			return err
+		}
 	}
-	ptr := createProtectedTimestampRecord(
-		ctx, cf.FlowCtx.Codec(), cf.spec.JobID, targets, tableHighWater,
-	)
-	if ptsEntries.ProtectedTimestampRecords == nil {
-		ptsEntries.ProtectedTimestampRecords = make(map[descpb.ID]*uuid.UUID)
+	return writeChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, ptsEntries, txn, cf.spec.JobID)
+}
+
+func (cf *changeFrontier) createPerTablePTSTarget(
+	tableID descpb.ID,
+) (changefeedbase.Targets, error) {
+	targets := changefeedbase.Targets{}
+	if cf.targets.Size > 0 {
+		if found, err := cf.targets.EachHavingTableID(tableID, func(target changefeedbase.Target) error {
+			targets.Add(target)
+			return nil
+		}); err != nil {
+			return changefeedbase.Targets{}, err
+		} else if !found {
+			return changefeedbase.Targets{}, errors.AssertionFailedf("attempted to create a per-table PTS record for table %d, but no target was found", tableID)
+		}
 	}
-	uuid := ptr.ID.GetUUID()
-	ptsEntries.ProtectedTimestampRecords[tableID] = &uuid
-	if err := pts.Protect(ctx, ptr); err != nil {
-		return err
+	if targets.Size != 1 {
+		return changefeedbase.Targets{}, errors.AssertionFailedf("expected 1 target, got %d", targets.Size)
 	}
-	return writeChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, ptsEntries, txn, cf.spec.JobID)
+	return targets, nil
 }
 
 func (cf *changeFrontier) advanceProtectedTimestamp(