jobs: only force jobs.MaybeGenerateForcedRetryableError in 23.1

pkg/ccl/backupccl/backup_job.go

@@ -262,7 +262,7 @@ func backup(
 					return nil
 				}
 				jobsprofiler.StorePerNodeProcessorProgressFraction(
-					ctx, execCtx.ExecCfg().InternalDB, job.ID(), prog)
+					ctx, execCtx.ExecCfg().InternalDB, job.ID(), prog, execCtx.ExecCfg().Settings.Version)
 			case <-ctx.Done():
 				return ctx.Err()
 			}
@@ -2095,7 +2095,7 @@ func (b *backupResumer) CollectProfile(ctx context.Context, execCtx interface{})
 		aggStatsCopy = b.mu.perNodeAggregatorStats.DeepCopy()
 	}()
 	return bulkutil.FlushTracingAggregatorStats(ctx, b.job.ID(),
-		p.ExecCfg().InternalDB, aggStatsCopy)
+		p.ExecCfg().InternalDB, aggStatsCopy, p.ExecCfg().Settings.Version)
 }
 
 func (b *backupResumer) deleteCheckpoint(

pkg/ccl/backupccl/backup_processor_planning.go

@@ -219,7 +219,7 @@ func distBackup(
 	defer close(progCh)
 	defer close(tracingAggCh)
 	execCfg := execCtx.ExecCfg()
-	jobsprofiler.StorePlanDiagram(ctx, execCfg.DistSQLSrv.Stopper, p, execCfg.InternalDB, jobID)
+	jobsprofiler.StorePlanDiagram(ctx, execCfg.DistSQLSrv.Stopper, p, execCfg.InternalDB, jobID, execCfg.Settings.Version)
 
 	// Copy the evalCtx, as dsp.Run() might change it.
 	evalCtxCopy := *evalCtx

pkg/ccl/backupccl/restore_job.go

@@ -2544,7 +2544,7 @@ func (r *restoreResumer) CollectProfile(ctx context.Context, execCtx interface{}
 		aggStatsCopy = r.mu.perNodeAggregatorStats.DeepCopy()
 	}()
 	return bulkutil.FlushTracingAggregatorStats(ctx, r.job.ID(),
-		p.ExecCfg().InternalDB, aggStatsCopy)
+		p.ExecCfg().InternalDB, aggStatsCopy, p.ExecCfg().Settings.Version)
 }
 
 // dropDescriptors implements the OnFailOrCancel logic.

pkg/ccl/backupccl/restore_processor_planning.go

@@ -317,7 +317,7 @@ func distRestore(
 		defer recv.Release()
 
 		execCfg := execCtx.ExecCfg()
-		jobsprofiler.StorePlanDiagram(ctx, execCfg.DistSQLSrv.Stopper, p, execCfg.InternalDB, md.jobID)
+		jobsprofiler.StorePlanDiagram(ctx, execCfg.DistSQLSrv.Stopper, p, execCfg.InternalDB, md.jobID, execCfg.Settings.Version)
 
 		// Copy the evalCtx, as dsp.Run() might change it.
 		evalCtxCopy := *evalCtx

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -303,7 +303,7 @@ func startDistChangefeed(
 			finishedSetupFn = func(flowinfra.Flow) { resultsCh <- tree.Datums(nil) }
 		}
 
-		jobsprofiler.StorePlanDiagram(ctx, execCfg.DistSQLSrv.Stopper, p, execCfg.InternalDB, jobID)
+		jobsprofiler.StorePlanDiagram(ctx, execCfg.DistSQLSrv.Stopper, p, execCfg.InternalDB, jobID, execCfg.Settings.Version)
 
 		// Copy the evalCtx, as dsp.Run() might change it.
 		evalCtxCopy := *evalCtx

pkg/ccl/streamingccl/replicationutils/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/ccl/streamingccl/replicationutils",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/clusterversion",
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/kv/kvpb",

pkg/ccl/streamingccl/replicationutils/utils.go

@@ -14,6 +14,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
@@ -174,9 +175,9 @@ func ReplicatedTimeFromProgress(p *jobspb.Progress) hlc.Timestamp {
 // LoadIngestionProgress loads the latest persisted stream ingestion progress.
 // The method returns nil if the progress does not exist yet.
 func LoadIngestionProgress(
-	ctx context.Context, db isql.DB, jobID jobspb.JobID,
+	ctx context.Context, db isql.DB, jobID jobspb.JobID, cv clusterversion.Handle,
 ) (*jobspb.StreamIngestionProgress, error) {
-	progress, err := jobs.LoadJobProgress(ctx, db, jobID)
+	progress, err := jobs.LoadJobProgress(ctx, db, jobID, cv)
 	if err != nil || progress == nil {
 		return nil, err
 	}
@@ -192,9 +193,9 @@ func LoadIngestionProgress(
 // LoadReplicationProgress loads the latest persisted stream replication progress.
 // The method returns nil if the progress does not exist yet.
 func LoadReplicationProgress(
-	ctx context.Context, db isql.DB, jobID jobspb.JobID,
+	ctx context.Context, db isql.DB, jobID jobspb.JobID, cv clusterversion.Handle,
 ) (*jobspb.StreamReplicationProgress, error) {
-	progress, err := jobs.LoadJobProgress(ctx, db, jobID)
+	progress, err := jobs.LoadJobProgress(ctx, db, jobID, cv)
 	if err != nil || progress == nil {
 		return nil, err
 	}

pkg/ccl/streamingccl/streamingest/BUILD.bazel

@@ -28,6 +28,7 @@ go_library(
         "//pkg/cloud",
         "//pkg/cloud/externalconn",
         "//pkg/cloud/externalconn/connectionpb",
+        "//pkg/clusterversion",
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/jobs/jobsprofiler",

pkg/ccl/streamingccl/streamingest/replication_execution_details.go

@@ -16,6 +16,7 @@ import (
 	"text/tabwriter"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -104,15 +105,19 @@ func constructSpanFrontierExecutionDetails(
 // - The snapshot of the frontier tracking how far each span has been replicated
 // up to.
 func generateSpanFrontierExecutionDetailFile(
-	ctx context.Context, execCfg *sql.ExecutorConfig, ingestionJobID jobspb.JobID, skipBehindBy bool,
+	ctx context.Context,
+	execCfg *sql.ExecutorConfig,
+	ingestionJobID jobspb.JobID,
+	skipBehindBy bool,
+	cv clusterversion.Handle,
 ) error {
 	return execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 		var sb bytes.Buffer
 		w := tabwriter.NewWriter(&sb, 0, 0, 1, ' ', tabwriter.TabIndent)
 
 		// Read the StreamIngestionPartitionSpecs to get a mapping from spans to
 		// their source and destination SQL instance IDs.
-		specs, err := jobs.ReadChunkedFileToJobInfo(ctx, replicationPartitionInfoFilename, txn, ingestionJobID)
+		specs, err := jobs.ReadChunkedFileToJobInfo(ctx, replicationPartitionInfoFilename, txn, ingestionJobID, cv)
 		if err != nil {
 			return err
 		}
@@ -124,7 +129,7 @@ func generateSpanFrontierExecutionDetailFile(
 
 		// Now, read the latest snapshot of the frontier that tells us what
 		// timestamp each span has been replicated up to.
-		frontierEntries, err := jobs.ReadChunkedFileToJobInfo(ctx, frontierEntriesFilename, txn, ingestionJobID)
+		frontierEntries, err := jobs.ReadChunkedFileToJobInfo(ctx, frontierEntriesFilename, txn, ingestionJobID, cv)
 		if err != nil {
 			return err
 		}
@@ -157,7 +162,7 @@ func generateSpanFrontierExecutionDetailFile(
 		if err := w.Flush(); err != nil {
 			return err
 		}
-		return jobs.WriteExecutionDetailFile(ctx, filename, sb.Bytes(), txn, ingestionJobID)
+		return jobs.WriteExecutionDetailFile(ctx, filename, sb.Bytes(), txn, ingestionJobID, cv)
 	})
 }
 
@@ -170,6 +175,7 @@ func persistStreamIngestionPartitionSpecs(
 	execCfg *sql.ExecutorConfig,
 	ingestionJobID jobspb.JobID,
 	streamIngestionSpecs map[base.SQLInstanceID]*execinfrapb.StreamIngestionDataSpec,
+	cv clusterversion.Handle,
 ) error {
 	err := execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 		specs := make([]*execinfrapb.StreamIngestionPartitionSpec, 0)
@@ -183,7 +189,7 @@ func persistStreamIngestionPartitionSpecs(
 		if err != nil {
 			return err
 		}
-		return jobs.WriteChunkedFileToJobInfo(ctx, replicationPartitionInfoFilename, specBytes, txn, ingestionJobID)
+		return jobs.WriteChunkedFileToJobInfo(ctx, replicationPartitionInfoFilename, specBytes, txn, ingestionJobID, cv)
 	})
 	if knobs := execCfg.StreamingTestingKnobs; knobs != nil && knobs.AfterPersistingPartitionSpecs != nil {
 		knobs.AfterPersistingPartitionSpecs()

pkg/ccl/streamingccl/streamingest/replication_execution_details_test.go

@@ -344,7 +344,7 @@ func TestEndToEndFrontierExecutionDetailFile(t *testing.T) {
 
 	ingestionJobID := jobspb.JobID(123)
 	require.NoError(t, persistStreamIngestionPartitionSpecs(ctx, &execCfg,
-		ingestionJobID, streamIngestionsSpecs))
+		ingestionJobID, streamIngestionsSpecs, execCfg.Settings.Version))
 
 	// Now, let's persist some frontier entries.
 	frontierEntries := execinfrapb.FrontierEntries{ResolvedSpans: []jobspb.ResolvedSpan{
@@ -369,9 +369,9 @@ func TestEndToEndFrontierExecutionDetailFile(t *testing.T) {
 	frontierBytes, err := protoutil.Marshal(&frontierEntries)
 	require.NoError(t, err)
 	require.NoError(t, execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
-		return jobs.WriteChunkedFileToJobInfo(ctx, frontierEntriesFilename, frontierBytes, txn, ingestionJobID)
+		return jobs.WriteChunkedFileToJobInfo(ctx, frontierEntriesFilename, frontierBytes, txn, ingestionJobID, execCfg.Settings.Version)
 	}))
-	require.NoError(t, generateSpanFrontierExecutionDetailFile(ctx, &execCfg, ingestionJobID, true /* skipBehindBy */))
+	require.NoError(t, generateSpanFrontierExecutionDetailFile(ctx, &execCfg, ingestionJobID, true /* skipBehindBy */, execCfg.Settings.Version))
 	files := listExecutionDetails(t, srv, ingestionJobID)
 	require.Len(t, files, 1)
 	data, err := checkExecutionDetails(t, srv, ingestionJobID, files[0])

pkg/ccl/streamingccl/streamingest/replication_stream_e2e_test.go

@@ -1114,12 +1114,14 @@ func TestLoadProducerAndIngestionProgress(t *testing.T) {
 	c.WaitUntilReplicatedTime(srcTime, jobspb.JobID(replicationJobID))
 
 	srcDB := c.SrcSysServer.ExecutorConfig().(sql.ExecutorConfig).InternalDB
-	producerProgress, err := replicationutils.LoadReplicationProgress(ctx, srcDB, jobspb.JobID(producerJobID))
+	producerProgress, err := replicationutils.LoadReplicationProgress(ctx, srcDB, jobspb.JobID(producerJobID),
+		c.SrcSysServer.ExecutorConfig().(sql.ExecutorConfig).Settings.Version)
 	require.NoError(t, err)
 	require.Equal(t, jobspb.StreamReplicationProgress_NOT_FINISHED, producerProgress.StreamIngestionStatus)
 
 	destDB := c.DestSysServer.ExecutorConfig().(sql.ExecutorConfig).InternalDB
-	ingestionProgress, err := replicationutils.LoadIngestionProgress(ctx, destDB, jobspb.JobID(replicationJobID))
+	ingestionProgress, err := replicationutils.LoadIngestionProgress(ctx, destDB, jobspb.JobID(replicationJobID),
+		c.DestSysServer.ExecutorConfig().(sql.ExecutorConfig).Settings.Version)
 	require.NoError(t, err)
 	require.Equal(t, jobspb.Replicating, ingestionProgress.ReplicationStatus)
 }

pkg/ccl/streamingccl/streamingest/stream_ingestion_dist.go

@@ -113,7 +113,7 @@ func startDistIngestion(
 		return errors.Wrap(err, "failed to update job progress")
 	}
 	jobsprofiler.StorePlanDiagram(ctx, execCtx.ExecCfg().DistSQLSrv.Stopper, planner.initialPlan, execCtx.ExecCfg().InternalDB,
-		ingestionJob.ID())
+		ingestionJob.ID(), execCtx.ExecCfg().Settings.Version)
 
 	replanOracle := sql.ReplanOnCustomFunc(
 		measurePlanChange,
@@ -463,7 +463,7 @@ func (p *replicationFlowPlanner) constructPlanGenerator(
 		if !p.createdInitialPlan() {
 			// Only persist the initial plan as it's the only plan that actually gets
 			// executed.
-			if err := persistStreamIngestionPartitionSpecs(ctx, execCtx.ExecCfg(), ingestionJobID, streamIngestionSpecs); err != nil {
+			if err := persistStreamIngestionPartitionSpecs(ctx, execCtx.ExecCfg(), ingestionJobID, streamIngestionSpecs, execCtx.ExecCfg().Settings.Version); err != nil {
 				return nil, nil, err
 			}
 		}

pkg/ccl/streamingccl/streamingest/stream_ingestion_frontier_processor.go

@@ -518,7 +518,7 @@ func (sf *streamIngestionFrontier) maybePersistFrontierEntries() error {
 	}
 
 	if err = sf.FlowCtx.Cfg.DB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
-		return jobs.WriteChunkedFileToJobInfo(ctx, frontierEntriesFilename, frontierBytes, txn, jobID)
+		return jobs.WriteChunkedFileToJobInfo(ctx, frontierEntriesFilename, frontierBytes, txn, jobID, sf.EvalCtx.Settings.Version)
 	}); err != nil {
 		return err
 	}

pkg/ccl/streamingccl/streamingest/stream_ingestion_job.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/streamingccl/replicationutils"
 	"github.com/cockroachdb/cockroach/pkg/ccl/streamingccl/streamclient"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobsprotectedts"
@@ -224,7 +225,7 @@ func ingestWithRetries(
 		}
 		status := redact.Sprintf("waiting before retrying error: %s", err)
 		updateRunningStatus(ctx, ingestionJob, jobspb.ReplicationError, status)
-		newReplicatedTime := loadReplicatedTime(ctx, execCtx.ExecCfg().InternalDB, ingestionJob)
+		newReplicatedTime := loadReplicatedTime(ctx, execCtx.ExecCfg().InternalDB, ingestionJob, execCtx.ExecCfg().Settings.Version)
 		if lastReplicatedTime.Less(newReplicatedTime) {
 			r.Reset()
 			lastReplicatedTime = newReplicatedTime
@@ -241,8 +242,10 @@ func ingestWithRetries(
 	return nil
 }
 
-func loadReplicatedTime(ctx context.Context, db isql.DB, ingestionJob *jobs.Job) hlc.Timestamp {
-	latestProgress, err := replicationutils.LoadIngestionProgress(ctx, db, ingestionJob.ID())
+func loadReplicatedTime(
+	ctx context.Context, db isql.DB, ingestionJob *jobs.Job, cv clusterversion.Handle,
+) hlc.Timestamp {
+	latestProgress, err := replicationutils.LoadIngestionProgress(ctx, db, ingestionJob.ID(), cv)
 	if err != nil {
 		log.Warningf(ctx, "error loading job progress: %s", err)
 		return hlc.Timestamp{}
@@ -556,11 +559,11 @@ func (s *streamIngestionResumer) CollectProfile(ctx context.Context, execCtx int
 
 	var combinedErr error
 	if err := bulkutil.FlushTracingAggregatorStats(ctx, s.job.ID(),
-		p.ExecCfg().InternalDB, aggStatsCopy); err != nil {
+		p.ExecCfg().InternalDB, aggStatsCopy, p.ExecCfg().Settings.Version); err != nil {
 		combinedErr = errors.CombineErrors(combinedErr, errors.Wrap(err, "failed to flush aggregator stats"))
 	}
 	if err := generateSpanFrontierExecutionDetailFile(ctx, p.ExecCfg(),
-		s.job.ID(), false /* skipBehindBy */); err != nil {
+		s.job.ID(), false /* skipBehindBy */, p.ExecCfg().Settings.Version); err != nil {
 		combinedErr = errors.CombineErrors(combinedErr, errors.Wrap(err, "failed to generate span frontier execution details"))
 	}
 

pkg/ccl/streamingccl/streamingest/stream_ingestion_processor.go

@@ -18,6 +18,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/streamingccl"
 	"github.com/cockroachdb/cockroach/pkg/ccl/streamingccl/replicationutils"
 	"github.com/cockroachdb/cockroach/pkg/ccl/streamingccl/streamclient"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv"
@@ -318,6 +319,7 @@ func newStreamIngestionDataProcessor(
 		cutoverProvider: &cutoverFromJobProgress{
 			jobID: jobspb.JobID(spec.JobID),
 			db:    flowCtx.Cfg.DB,
+			cv:    flowCtx.Cfg.Settings.Version,
 		},
 		buffer:           &streamIngestionBuffer{},
 		cutoverCh:        make(chan struct{}),
@@ -1245,10 +1247,11 @@ type cutoverProvider interface {
 type cutoverFromJobProgress struct {
 	db    isql.DB
 	jobID jobspb.JobID
+	cv    clusterversion.Handle
 }
 
 func (c *cutoverFromJobProgress) cutoverReached(ctx context.Context) (bool, error) {
-	ingestionProgress, err := replicationutils.LoadIngestionProgress(ctx, c.db, c.jobID)
+	ingestionProgress, err := replicationutils.LoadIngestionProgress(ctx, c.db, c.jobID, c.cv)
 	if err != nil {
 		return false, err
 	}

pkg/ccl/streamingccl/streamproducer/producer_job.go

@@ -99,7 +99,7 @@ func (p *producerJobResumer) Resume(ctx context.Context, execCtx interface{}) er
 		case <-p.timer.Ch():
 			p.timer.MarkRead()
 			p.timer.Reset(streamingccl.StreamReplicationStreamLivenessTrackFrequency.Get(execCfg.SV()))
-			progress, err := replicationutils.LoadReplicationProgress(ctx, execCfg.InternalDB, p.job.ID())
+			progress, err := replicationutils.LoadReplicationProgress(ctx, execCfg.InternalDB, p.job.ID(), execCfg.Settings.Version)
 			if knobs := execCfg.StreamingTestingKnobs; knobs != nil && knobs.AfterResumerJobLoad != nil {
 				err = knobs.AfterResumerJobLoad(err)
 			}

pkg/jobs/adopt.go

@@ -87,7 +87,7 @@ func (r *Registry) maybeDumpTrace(resumerCtx context.Context, resumer Resumer, j
 		r.ID().String(), timeutil.Now().Format("20060102_150405.00"))
 	td := jobspb.TraceData{CollectedSpans: sp.GetConfiguredRecording()}
 	if err := r.db.Txn(dumpCtx, func(ctx context.Context, txn isql.Txn) error {
-		return WriteProtobinExecutionDetailFile(dumpCtx, resumerTraceFilename, &td, txn, jobID)
+		return WriteProtobinExecutionDetailFile(dumpCtx, resumerTraceFilename, &td, txn, jobID, r.settings.Version)
 	}); err != nil {
 		log.Warning(dumpCtx, "failed to write trace on resumer trace file")
 	}