WIP to remove bucketing, need to get test coverage back

Author: aerfrei

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -1944,15 +1944,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)
@@ -1963,28 +1955,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 {
@@ -1996,22 +1968,9 @@ 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 updated, err := cf.advancePerTableProtectedTimestampRecord(ctx, ptsEntries, tableID, tableHighWater, pts); err != nil {
-				return hlc.Timestamp{}, false, err
+				return false, err
 			} else if updated {
 				updatedPerTablePTS = true
 			}
@@ -2021,41 +1980,25 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 		}
 	}
 
-	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)
+	if cf.knobs.ManagePTSError != nil {
+		return false, cf.knobs.ManagePTSError()
 	}
-	return nil
+
+	return updatedPerTablePTS, nil
 }
 
 func (cf *changeFrontier) advancePerTableProtectedTimestampRecord(

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/backup/backupresolver"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdceval"
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcprogresspb"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedvalidators"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/checkpoint"
@@ -320,15 +321,44 @@ func changefeedPlanHook(
 			recordPTSMetricsTime := sliMetrics.Timers.PTSCreate.Start()
 
 			var ptr *ptpb.Record
+			var ptrs []*ptpb.Record // *cdcprogresspb.ProtectedTimestampRecords
+			var jobInfoPTRs *cdcprogresspb.ProtectedTimestampRecords
 			codec := p.ExecCfg().Codec
-			ptr = createProtectedTimestampRecord(
-				ctx,
-				codec,
-				jobID,
-				targets,
-				details.StatementTime,
-			)
-			progress.GetChangefeed().ProtectedTimestampRecord = ptr.ID.GetUUID()
+
+			// We do not yet have the progress config here, so we need to check the settings directly.
+			perTableTrackingEnabled := changefeedbase.TrackPerTableProgress.Get(&p.ExecCfg().Settings.SV)
+			perTableProtectedTimestampsEnabled := changefeedbase.PerTableProtectedTimestamps.Get(&p.ExecCfg().Settings.SV)
+			writingMultiplePTS := perTableTrackingEnabled && perTableProtectedTimestampsEnabled
+			if writingMultiplePTS {
+				protectedTimestampRecords := make(map[descpb.ID]*uuid.UUID)
+				targets.EachTarget(func(target changefeedbase.Target) error {
+					ptsTargets := changefeedbase.Targets{}
+					ptsTargets.Add(target)
+					ptsRecord := createProtectedTimestampRecord(
+						ctx,
+						codec,
+						jobID,
+						ptsTargets,
+						details.StatementTime,
+					)
+					ptrs = append(ptrs, ptsRecord)
+					uuid := ptsRecord.ID.GetUUID()
+					protectedTimestampRecords[target.DescID] = &uuid
+					return nil
+				})
+				jobInfoPTRs = &cdcprogresspb.ProtectedTimestampRecords{
+					ProtectedTimestampRecords: protectedTimestampRecords,
+				}
+			} else {
+				ptr = createProtectedTimestampRecord(
+					ctx,
+					codec,
+					jobID,
+					targets,
+					details.StatementTime,
+				)
+				progress.GetChangefeed().ProtectedTimestampRecord = ptr.ID.GetUUID()
+			}
 
 			jr.Progress = *progress.GetChangefeed()
 
@@ -348,6 +378,19 @@ func changefeedPlanHook(
 						return err
 					}
 				}
+				if len(ptrs) > 0 {
+					for _, ptr := range ptrs {
+						pts := p.ExecCfg().ProtectedTimestampProvider.WithTxn(p.InternalSQLTxn())
+						if err := pts.Protect(ctx, ptr); err != nil {
+							return err
+						}
+					}
+					if err := writeChangefeedJobInfo(
+						ctx, perTableProtectedTimestampsFilename, jobInfoPTRs, p.InternalSQLTxn(), jobID,
+					); err != nil {
+						return err
+					}
+				}
 
 				select {
 				case <-ctx.Done():
@@ -364,7 +407,23 @@ func changefeedPlanHook(
 					return err
 				}
 				if ptr != nil {
-					return p.ExecCfg().ProtectedTimestampProvider.WithTxn(txn).Protect(ctx, ptr)
+					err := p.ExecCfg().ProtectedTimestampProvider.WithTxn(txn).Protect(ctx, ptr)
+					if err != nil {
+						return err
+					}
+				}
+				if len(ptrs) > 0 {
+					pts := p.ExecCfg().ProtectedTimestampProvider.WithTxn(txn)
+					for _, ptr := range ptrs {
+						if err := pts.Protect(ctx, ptr); err != nil {
+							return err
+						}
+					}
+					if err := writeChangefeedJobInfo(
+						ctx, perTableProtectedTimestampsFilename, jobInfoPTRs, txn, jobID,
+					); err != nil {
+						return err
+					}
 				}
 				return nil
 			}); err != nil {

pkg/ccl/changefeedccl/changefeed_test.go

@@ -36,6 +36,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/build"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdceval"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcevent"
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcprogresspb"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdctest"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/checkpoint"
@@ -65,6 +66,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
@@ -12084,15 +12086,47 @@ func TestChangefeedProtectedTimestampUpdate(t *testing.T) {
 			return errors.New("waiting for high watermark to advance")
 		}
 		testutils.SucceedsSoon(t, checkHWM)
+		var tableID int
+		sqlDB.QueryRow(t, `SELECT table_id FROM crdb_internal.tables WHERE name = 'foo' AND database_name = 'd'`).Scan(&tableID)
+		execCfg := s.Server.ExecutorConfig().(sql.ExecutorConfig)
+
+		getTimestampFromPTSRecord := func(ptsRecordID uuid.UUID) hlc.Timestamp {
+			ptsQry := fmt.Sprintf(`SELECT ts FROM system.protected_ts_records WHERE id = '%s'`, ptsRecordID)
+			var tsStr string
+			sqlDB.QueryRow(t, ptsQry).Scan(&tsStr)
+			ts, err := hlc.ParseHLC(tsStr)
+			require.NoError(t, err)
+			return ts
+		}
+		getPerTablePTS := func() hlc.Timestamp {
+			var ts hlc.Timestamp
+			err := execCfg.InternalDB.Txn(context.Background(), func(ctx context.Context, txn isql.Txn) error {
+				var ptsEntries cdcprogresspb.ProtectedTimestampRecords
+				if err := readChangefeedJobInfo(
+					ctx, perTableProtectedTimestampsFilename, &ptsEntries, txn, eFeed.JobID(),
+				); err != nil {
+					return err
+				}
+				ptsRecordID := ptsEntries.ProtectedTimestampRecords[descpb.ID(tableID)]
+				ts = getTimestampFromPTSRecord(*ptsRecordID)
+				return nil
+			})
+			require.NoError(t, err)
+			return ts
+		}
+		perTablePTSEnabled := changefeedbase.PerTableProtectedTimestamps.Get(&s.Server.ClusterSettings().SV)
+		perTableProgressEnabled := changefeedbase.TrackPerTableProgress.Get(&s.Server.ClusterSettings().SV)
+		getPTS := func() hlc.Timestamp {
+			if perTablePTSEnabled && perTableProgressEnabled {
+				return getPerTablePTS()
+			}
 
-		// Get the PTS of this feed.
-		p, err := eFeed.Progress()
-		require.NoError(t, err)
+			p, err := eFeed.Progress()
+			require.NoError(t, err)
+			return getTimestampFromPTSRecord(p.ProtectedTimestampRecord)
+		}
 
-		ptsQry := fmt.Sprintf(`SELECT ts FROM system.protected_ts_records WHERE id = '%s'`, p.ProtectedTimestampRecord)
-		var ts, ts2 string
-		sqlDB.QueryRow(t, ptsQry).Scan(&ts)
-		require.NoError(t, err)
+		ts := getPTS()
 
 		// Force the changefeed to restart.
 		require.NoError(t, eFeed.Pause())
@@ -12102,8 +12136,7 @@ func TestChangefeedProtectedTimestampUpdate(t *testing.T) {
 		testutils.SucceedsSoon(t, checkHWM)
 
 		// Check that the PTS was not updated after the resume.
-		sqlDB.QueryRow(t, ptsQry).Scan(&ts2)
-		require.NoError(t, err)
+		ts2 := getPTS()
 		require.Equal(t, ts, ts2)
 
 		// Lower the PTS lag and check that it has been updated.
@@ -12115,9 +12148,8 @@ func TestChangefeedProtectedTimestampUpdate(t *testing.T) {
 		testutils.SucceedsSoon(t, checkHWM)
 		testutils.SucceedsSoon(t, checkHWM)
 
-		sqlDB.QueryRow(t, ptsQry).Scan(&ts2)
-		require.NoError(t, err)
-		require.Less(t, ts, ts2)
+		ts2 = getPTS()
+		require.True(t, ts.Less(ts2))
 
 		managePTSCount, _ = metrics.AggMetrics.Timers.PTSManage.WindowedSnapshot().Total()
 		managePTSErrorCount, _ = metrics.AggMetrics.Timers.PTSManageError.WindowedSnapshot().Total()