changefeedccl: fix timestamp for desc fetches during planning

When a changefeed is planned and executed, there are several places
where target table descriptors are fetched. With a db-level changefeed,
the set of tables can change during changefeed startup. Now, the
timestamp is set to the schema timestamp for retrieving table
descriptors.

Adding a schema_ts to the protobuf for ChangeAggregatorSpec and
ChangeFrontierSpec.

Fixes: #154549
Epic: CRDB-1421

Release note: None

Author: log-head

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -448,7 +448,11 @@ func generateAndValidateNewTargets(
 		return nil, nil, hlc.Timestamp{}, nil, err
 	}
 
-	prevTargets, err := AllTargets(ctx, prevDetails, p.ExecCfg())
+	targetTS := prevProgress.GetHighWater()
+	if targetTS == nil || targetTS.IsEmpty() {
+		targetTS = &prevDetails.StatementTime
+	}
+	prevTargets, err := AllTargets(ctx, prevDetails, p.ExecCfg(), *targetTS)
 	if err != nil {
 		return nil, nil, hlc.Timestamp{}, nil, err
 	}

pkg/ccl/changefeedccl/changefeed.go

@@ -62,7 +62,10 @@ func makeChangefeedConfigFromJobDetails(
 // from the statement time name map in old protos
 // or the TargetSpecifications in new ones.
 func AllTargets(
-	ctx context.Context, cd jobspb.ChangefeedDetails, execCfg *sql.ExecutorConfig,
+	ctx context.Context,
+	cd jobspb.ChangefeedDetails,
+	execCfg *sql.ExecutorConfig,
+	timestamp hlc.Timestamp,
 ) (changefeedbase.Targets, error) {
 	targets := changefeedbase.Targets{}
 	var err error
@@ -76,7 +79,7 @@ func AllTargets(
 					if len(cd.TargetSpecifications) > 1 {
 						return changefeedbase.Targets{}, errors.AssertionFailedf("database-level changefeed is not supported with multiple targets")
 					}
-					targets, err = getTargetsFromDatabaseSpec(ctx, ts, execCfg)
+					targets, err = getTargetsFromDatabaseSpec(ctx, ts, execCfg, timestamp)
 					if err != nil {
 						return changefeedbase.Targets{}, err
 					}
@@ -110,11 +113,17 @@ func AllTargets(
 }
 
 func getTargetsFromDatabaseSpec(
-	ctx context.Context, ts jobspb.ChangefeedTargetSpecification, execCfg *sql.ExecutorConfig,
+	ctx context.Context,
+	ts jobspb.ChangefeedTargetSpecification,
+	execCfg *sql.ExecutorConfig,
+	timestamp hlc.Timestamp,
 ) (targets changefeedbase.Targets, err error) {
 	err = sql.DescsTxn(ctx, execCfg, func(
 		ctx context.Context, txn isql.Txn, descs *descs.Collection,
 	) error {
+		if err := txn.KV().SetFixedTimestamp(ctx, timestamp); err != nil {
+			return errors.Wrapf(err, "setting timestamp for table descriptor fetch")
+		}
 		databaseDescriptor, err := descs.ByIDWithLeased(txn.KV()).Get().Database(ctx, ts.DescID)
 		if err != nil {
 			return err

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -56,35 +56,16 @@ const (
 	changeFrontierProcName   = `changefntr`
 )
 
-// distChangefeedFlow plans and runs a distributed changefeed.
-//
-// One or more ChangeAggregator processors watch table data for changes. These
-// transform the changed kvs into changed rows and either emit them to a sink
-// (such as kafka) or, if there is no sink, forward them in columns 1,2,3 (where
-// they will be eventually returned directly via pgwire). In either case,
-// periodically a span will become resolved as of some timestamp, meaning that
-// no new rows will ever be emitted at or below that timestamp. These span-level
-// resolved timestamps are emitted as a marshaled `jobspb.ResolvedSpan` proto in
-// column 0.
-//
-// The flow will always have exactly one ChangeFrontier processor which all the
-// ChangeAggregators feed into. It collects all span-level resolved timestamps
-// and aggregates them into a changefeed-level resolved timestamp, which is the
-// minimum of the span-level resolved timestamps. This changefeed-level resolved
-// timestamp is emitted into the changefeed sink (or returned to the gateway if
-// there is no sink) whenever it advances. ChangeFrontier also updates the
-// progress of the changefeed's corresponding system job.
-func distChangefeedFlow(
+// computeDistChangefeedTimestamps computes the initialHighWater and schemaTS
+// for a changefeed run, and mutates localState.progress when appropriate
+// (e.g., to set the high-water if initial scan should be skipped). It also
+// invokes testing knobs that observe the initial high-water.
+func computeDistChangefeedTimestamps(
 	ctx context.Context,
 	execCtx sql.JobExecContext,
-	jobID jobspb.JobID,
 	details jobspb.ChangefeedDetails,
-	description string,
 	localState *cachedState,
-	resultsCh chan<- tree.Datums,
-	onTracingEvent func(ctx context.Context, meta *execinfrapb.TracingAggregatorEvents),
-	targets changefeedbase.Targets,
-) error {
+) (initialHighWater hlc.Timestamp, schemaTS hlc.Timestamp, _ error) {
 	opts := changefeedbase.MakeStatementOptions(details.Opts)
 	progress := localState.progress
 
@@ -100,44 +81,38 @@ func distChangefeedFlow(
 		// cursor but we have a request to not have an initial scan.
 		initialScanType, err := opts.GetInitialScanType()
 		if err != nil {
-			return err
+			return hlc.Timestamp{}, hlc.Timestamp{}, err
 		}
 		if noHighWater && initialScanType == changefeedbase.NoInitialScan {
 			// If there is a cursor, the statement time has already been set to it.
 			progress.Progress = &jobspb.Progress_HighWater{HighWater: &details.StatementTime}
 		}
 	}
 
-	var initialHighWater hlc.Timestamp
-	schemaTS := details.StatementTime
-	{
-		if h := progress.GetHighWater(); h != nil && !h.IsEmpty() {
-			initialHighWater = *h
-			// If we have a high-water set, use it to compute the spans, since the
-			// ones at the statement time may have been garbage collected by now.
-			schemaTS = initialHighWater
-		}
-
-		// We want to fetch the target spans as of the timestamp following the
-		// highwater unless the highwater corresponds to a timestamp of an initial
-		// scan. This logic is irritatingly complex but extremely important. Namely,
-		// we may be here because the schema changed at the current resolved
-		// timestamp. However, an initial scan should be performed at exactly the
-		// timestamp specified; initial scans can be created at the timestamp of a
-		// schema change and thus should see the side-effect of the schema change.
-		isRestartAfterCheckpointOrNoInitialScan := progress.GetHighWater() != nil
-		if isRestartAfterCheckpointOrNoInitialScan {
-			schemaTS = schemaTS.Next()
-		}
+	schemaTS = details.StatementTime
+	if h := progress.GetHighWater(); h != nil && !h.IsEmpty() {
+		initialHighWater = *h
+		// If we have a high-water set, use it to compute the spans, since the
+		// ones at the statement time may have been garbage collected by now.
+		schemaTS = initialHighWater
+	}
+	// We want to fetch the target spans as of the timestamp following the
+	// highwater unless the highwater corresponds to a timestamp of an initial
+	// scan. This logic is irritatingly complex but extremely important. Namely,
+	// we may be here because the schema changed at the current resolved
+	// timestamp. However, an initial scan should be performed at exactly the
+	// timestamp specified; initial scans can be created at the timestamp of a
+	// schema change and thus should see the side-effect of the schema change.
+	if progress.GetHighWater() != nil {
+		schemaTS = schemaTS.Next()
 	}
 
 	if knobs, ok := execCtx.ExecCfg().DistSQLSrv.TestingKnobs.Changefeed.(*TestingKnobs); ok {
 		if knobs != nil && knobs.StartDistChangefeedInitialHighwater != nil {
 			knobs.StartDistChangefeedInitialHighwater(ctx, initialHighWater)
 		}
 	}
-	return startDistChangefeed(
-		ctx, execCtx, jobID, schemaTS, details, description, initialHighWater, localState, resultsCh, onTracingEvent, targets)
+	return initialHighWater, schemaTS, nil
 }
 
 func fetchTableDescriptors(
@@ -225,7 +200,24 @@ func fetchSpansForTables(
 		sd, tableDescs[0], initialHighwater, target, sc)
 }
 
-// startDistChangefeed starts distributed changefeed execution.
+// startDistChangefeed plans and runs a distributed changefeed.
+//
+// One or more ChangeAggregator processors watch table data for changes. These
+// transform the changed kvs into changed rows and either emit them to a sink
+// (such as kafka) or, if there is no sink, forward them in columns 1,2,3 (where
+// they will be eventually returned directly via pgwire). In either case,
+// periodically a span will become resolved as of some timestamp, meaning that
+// no new rows will ever be emitted at or below that timestamp. These span-level
+// resolved timestamps are emitted as a marshaled `jobspb.ResolvedSpan` proto in
+// column 0.
+//
+// The flow will always have exactly one ChangeFrontier processor which all the
+// ChangeAggregators feed into. It collects all span-level resolved timestamps
+// and aggregates them into a changefeed-level resolved timestamp, which is the
+// minimum of the span-level resolved timestamps. This changefeed-level resolved
+// timestamp is emitted into the changefeed sink (or returned to the gateway if
+// there is no sink) whenever it advances. ChangeFrontier also updates the
+// progress of the changefeed's corresponding system job.
 func startDistChangefeed(
 	ctx context.Context,
 	execCtx sql.JobExecContext,
@@ -270,7 +262,7 @@ func startDistChangefeed(
 		}
 	}
 	p, planCtx, err := makePlan(execCtx, jobID, details, description, initialHighWater,
-		trackedSpans, spanLevelCheckpoint, localState.drainingNodes)(ctx, dsp)
+		trackedSpans, spanLevelCheckpoint, localState.drainingNodes, schemaTS)(ctx, dsp)
 	if err != nil {
 		return err
 	}
@@ -395,6 +387,7 @@ func makePlan(
 	trackedSpans []roachpb.Span,
 	spanLevelCheckpoint *jobspb.TimestampSpansMap,
 	drainingNodes []roachpb.NodeID,
+	schemaTS hlc.Timestamp,
 ) func(context.Context, *sql.DistSQLPlanner) (*sql.PhysicalPlan, *sql.PlanningCtx, error) {
 	return func(ctx context.Context, dsp *sql.DistSQLPlanner) (*sql.PhysicalPlan, *sql.PlanningCtx, error) {
 		sv := &execCtx.ExecCfg().Settings.SV
@@ -535,6 +528,7 @@ func makePlan(
 				Description:         description,
 				ProgressConfig:      progressConfig,
 				ResolvedSpans:       resolvedSpans,
+				SchemaTS:            &schemaTS,
 			}
 		}
 
@@ -551,6 +545,7 @@ func makePlan(
 			Description:         description,
 			ProgressConfig:      progressConfig,
 			ResolvedSpans:       resolvedSpans,
+			SchemaTS:            &schemaTS,
 		}
 
 		if haveKnobs && maybeCfKnobs.OnDistflowSpec != nil {

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -378,7 +378,8 @@ func (ca *changeAggregator) Start(ctx context.Context) {
 	if ca.knobs.OverrideExecCfg != nil {
 		execCfg = ca.knobs.OverrideExecCfg(execCfg)
 	}
-	ca.targets, err = AllTargets(ctx, ca.spec.Feed, execCfg)
+	targetTS := ca.spec.GetSchemaTS()
+	ca.targets, err = AllTargets(ctx, ca.spec.Feed, execCfg, targetTS)
 	if err != nil {
 		log.Changefeed.Warningf(ca.Ctx(), "moving to draining due to error getting targets: %v", err)
 		ca.MoveToDraining(err)
@@ -1352,7 +1353,8 @@ func newChangeFrontierProcessor(
 	if cf.knobs.OverrideExecCfg != nil {
 		execCfg = cf.knobs.OverrideExecCfg(execCfg)
 	}
-	targets, err := AllTargets(ctx, spec.Feed, execCfg)
+	targetTS := cf.spec.GetSchemaTS()
+	targets, err := AllTargets(ctx, cf.spec.Feed, execCfg, targetTS)
 	if err != nil {
 		return nil, err
 	}

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -534,7 +534,15 @@ func coreChangefeed(
 			knobs.BeforeDistChangefeed()
 		}
 
-		err := distChangefeedFlow(ctx, p, 0 /* jobID */, details, description, localState, resultsCh, nil, targets)
+		initialHighWater, schemaTS, err := computeDistChangefeedTimestamps(ctx, p, details, localState)
+		if err != nil {
+			return err
+		}
+		maybeCfKnobs, haveKnobs := p.ExecCfg().DistSQLSrv.TestingKnobs.Changefeed.(*TestingKnobs)
+		if haveKnobs && maybeCfKnobs.AfterComputeDistChangefeedTimestamps != nil {
+			maybeCfKnobs.AfterComputeDistChangefeedTimestamps(ctx)
+		}
+		err = startDistChangefeed(ctx, p, 0 /* jobID */, schemaTS, details, description, initialHighWater, localState, resultsCh, nil, targets)
 		if err == nil {
 			log.Changefeed.Infof(ctx, "core changefeed completed with no error")
 			return nil
@@ -744,7 +752,7 @@ func createChangefeedJobRecord(
 		return nil, changefeedbase.Targets{}, errors.AssertionFailedf("unknown changefeed level: %s", changefeedStmt.Level)
 	}
 
-	targets, err := AllTargets(ctx, details, p.ExecCfg())
+	targets, err := AllTargets(ctx, details, p.ExecCfg(), statementTime)
 	if err != nil {
 		return nil, changefeedbase.Targets{}, err
 	}
@@ -1757,13 +1765,22 @@ func (b *changefeedResumer) resumeWithRetries(
 
 			confPoller := make(chan struct{})
 			g := ctxgroup.WithContext(ctx)
-			targets, err := AllTargets(ctx, details, execCfg)
+			initialHighWater, schemaTS, err := computeDistChangefeedTimestamps(ctx, jobExec, details, localState)
+			if err != nil {
+				return err
+			}
+			maybeCfKnobs, haveKnobs := execCfg.DistSQLSrv.TestingKnobs.Changefeed.(*TestingKnobs)
+			if haveKnobs && maybeCfKnobs.AfterComputeDistChangefeedTimestamps != nil {
+				maybeCfKnobs.AfterComputeDistChangefeedTimestamps(ctx)
+			}
+			targets, err := AllTargets(ctx, details, execCfg, schemaTS)
 			if err != nil {
 				return err
 			}
 			g.GoCtx(func(ctx context.Context) error {
 				defer close(confPoller)
-				return distChangefeedFlow(ctx, jobExec, jobID, details, description, localState, startedCh, onTracingEvent, targets)
+				return startDistChangefeed(ctx, jobExec, jobID, schemaTS, details, description,
+					initialHighWater, localState, startedCh, onTracingEvent, targets)
 			})
 			g.GoCtx(func(ctx context.Context) error {
 				t := time.NewTicker(15 * time.Second)
@@ -2012,7 +2029,12 @@ func (b *changefeedResumer) OnFailOrCancel(
 
 	var numTargets uint
 	if b.job != nil {
-		targets, err := AllTargets(ctx, b.job.Details().(jobspb.ChangefeedDetails), execCfg)
+		targetsTS := progress.GetHighWater()
+		if targetsTS == nil || targetsTS.IsEmpty() {
+			details := b.job.Details().(jobspb.ChangefeedDetails)
+			targetsTS = &details.StatementTime
+		}
+		targets, err := AllTargets(ctx, b.job.Details().(jobspb.ChangefeedDetails), execCfg, *targetsTS)
 		if err != nil {
 			return err
 		}