Merge #40288

40288: jobs: make new index backfill status tracking bwd compatible r=spaskob a=spaskob

Currently if a new node participates in a cluster with old version nodes the index backfill
will be busy waiting since the different versions use different ways of reporting status.
We detect this situation and revert to the old status reporting style if necessary.

Touches #36601.

Release note: None

Co-authored-by: Spas Bojanov <spas@cockroachlabs.com>

pkg/sql/backfill.go

@@ -704,6 +704,7 @@ func (sc *SchemaChanger) distBackfill(
 	filter backfill.MutationFilter,
 	targetSpans []roachpb.Span,
 ) error {
+	inMemoryStatusEnabled := sc.execCfg.Settings.Version.IsActive(cluster.VersionAtomicChangeReplicasTrigger)
 	duration := checkpointInterval
 	if sc.testingKnobs.WriteCheckpointInterval > 0 {
 		duration = sc.testingKnobs.WriteCheckpointInterval
@@ -879,6 +880,24 @@ func (sc *SchemaChanger) distBackfill(
 			}); err != nil {
 				return err
 			}
+			if !inMemoryStatusEnabled {
+				var resumeSpans []roachpb.Span
+				// There is a worker node of older version that will communicate
+				// its done work by writing to the jobs table.
+				// In this case we intersect todoSpans with what the old node(s)
+				// have set in the jobs table not to overwrite their done work.
+				if err := sc.db.Txn(ctx, func(ctx context.Context, txn *client.Txn) error {
+					var err error
+					resumeSpans, _, _, err = distsqlrun.GetResumeSpans(
+						ctx, sc.jobRegistry, txn, sc.tableID, sc.mutationID, filter)
+					return err
+				}); err != nil {
+					return err
+				}
+				// A \intersect B = A - (A - B)
+				todoSpans = roachpb.SubtractSpans(todoSpans, roachpb.SubtractSpans(todoSpans, resumeSpans))
+
+			}
 			// Record what is left to do for the job.
 			// TODO(spaskob): Execute this at a regular cadence.
 			if err := sc.db.Txn(ctx, func(ctx context.Context, txn *client.Txn) error {

pkg/sql/distsqlrun/backfiller.go

@@ -18,6 +18,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/backfill"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsqlpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
@@ -134,6 +135,19 @@ func (b *backfiller) doRun(ctx context.Context) *distsqlpb.ProducerMetadata {
 	if err != nil {
 		return &distsqlpb.ProducerMetadata{Err: err}
 	}
+	if !b.flowCtx.Cfg.Settings.Version.IsActive(cluster.VersionAtomicChangeReplicasTrigger) {
+		// There is a node of older version which could be the coordinator.
+		// So we communicate the finished work by writing to the jobs row.
+		err = WriteResumeSpan(ctx,
+			b.flowCtx.Cfg.DB,
+			b.spec.Table.ID,
+			b.spec.Table.Mutations[0].MutationID,
+			b.filter,
+			finishedSpans,
+			b.flowCtx.Cfg.JobRegistry,
+		)
+		return &distsqlpb.ProducerMetadata{Err: err}
+	}
 	var prog distsqlpb.RemoteProducerMetadata_BulkProcessorProgress
 	prog.CompletedSpans = append(prog.CompletedSpans, finishedSpans...)
 	return &distsqlpb.ProducerMetadata{BulkProcessorProgress: &prog}