sql/schemachanger: fix index creation failures after pause/resume

Previously, when the index backfiller was tracking completed spans, the
final chunk of a span would cause an update to be emitted that
incorrectly indicated the entire span was complete.  While this was not
an issue with a single ingest goroutine, it caused problems with the
introduction of multiple ingest goroutines. As a result, if the job was
paused and resumed after this incorrect progress update, the backfiller
could skip rows, leading to validation errors.  To address this, this
patch ensures that progress updates for the final chunk of a span
correctly report only the work completed in that chunk.

Fixes: #153522

Release note (bug fix): Addressed a bug where index creation could fail
due to validation errors if the schema change was retried or
paused/resumed during the backfill.

Author: fqazi

pkg/sql/backfill/backfill.go

@@ -948,19 +948,21 @@ func (ib *IndexBackfiller) init(
 // that needs to be freed once the returned IndexEntry slice is freed. This is
 // returned for the successful and failure cases. It is the callers responsibility
 // to clear the associated bound account when appropriate.
+// A non-nil resumeKey is returned when there is still work to be done in this
+// span (sp.Key < resumeKey < sp.EndKey), which happens if the entire span does
+// not fit within the batch size.
 func (ib *IndexBackfiller) BuildIndexEntriesChunk(
 	ctx context.Context,
 	txn *kv.Txn,
 	tableDesc catalog.TableDescriptor,
 	sp roachpb.Span,
 	chunkSize int64,
 	traceKV bool,
-) ([]rowenc.IndexEntry, roachpb.Key, int64, error) {
+) (entries []rowenc.IndexEntry, resumeKey roachpb.Key, memUsedPerChunk int64, err error) {
 	// This ought to be chunkSize but in most tests we are actually building smaller
 	// indexes so use a smaller value.
 	const initBufferSize = 1000
 	const sizeOfIndexEntry = int64(unsafe.Sizeof(rowenc.IndexEntry{}))
-	var memUsedPerChunk int64
 
 	indexEntriesInChunkInitialBufferSize :=
 		sizeOfIndexEntry * initBufferSize * int64(len(ib.added))
@@ -969,7 +971,7 @@ func (ib *IndexBackfiller) BuildIndexEntriesChunk(
 			"failed to initialize empty buffer to store the index entries of all rows in the chunk")
 	}
 	memUsedPerChunk += indexEntriesInChunkInitialBufferSize
-	entries := make([]rowenc.IndexEntry, 0, initBufferSize*int64(len(ib.added)))
+	entries = make([]rowenc.IndexEntry, 0, initBufferSize*int64(len(ib.added)))
 
 	var fetcherCols []descpb.ColumnID
 	for i, c := range ib.cols {
@@ -1202,7 +1204,6 @@ func (ib *IndexBackfiller) BuildIndexEntriesChunk(
 	ib.ShrinkBoundAccount(ctx, shrinkSize)
 	memUsedPerChunk -= shrinkSize
 
-	var resumeKey roachpb.Key
 	if fetcher.Key() != nil {
 		resumeKey = make(roachpb.Key, len(fetcher.Key()))
 		copy(resumeKey, fetcher.Key())

pkg/sql/index_backfiller.go

@@ -106,7 +106,7 @@ func (ib *IndexBackfillPlanner) BackfillIndexes(
 
 		knobs := &ib.execCfg.DistSQLSrv.TestingKnobs
 		if knobs.RunBeforeIndexBackfillProgressUpdate != nil {
-			knobs.RunBeforeIndexBackfillProgressUpdate(ctx, progress.CompletedSpans)
+			knobs.RunBeforeIndexBackfillProgressUpdate(ctx, meta.BulkProcessorProgress.CompletedSpans)
 		}
 		return tracker.SetBackfillProgress(ctx, progress)
 	}

pkg/sql/indexbackfiller_test.go

@@ -37,6 +37,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scexec"
+	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scop"
+	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scplan"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/catid"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/idxtype"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -581,7 +583,8 @@ func TestIndexBackfillerResumePreservesProgress(t *testing.T) {
 
 	ctx := context.Background()
 	backfillProgressCompletedCh := make(chan []roachpb.Span)
-	const numSpans = 100
+	const numRows = 100
+	const numSpans = 20
 	var isBlockingBackfillProgress atomic.Bool
 	isBlockingBackfillProgress.Store(true)
 
@@ -610,22 +613,35 @@ func TestIndexBackfillerResumePreservesProgress(t *testing.T) {
 					}
 					return nil
 				},
+				AfterStage: func(p scplan.Plan, stageIdx int) error {
+					if p.Stages[stageIdx].Type() != scop.BackfillType || !isBlockingBackfillProgress.Load() {
+						return nil
+					}
+					isBlockingBackfillProgress.Store(false)
+					close(backfillProgressCompletedCh)
+					return nil
+				},
 			},
 		},
 	})
 	defer tc.Stopper().Stop(ctx)
 
 	_, err := db.Exec(`SET CLUSTER SETTING bulkio.index_backfill.batch_size = 10`)
 	require.NoError(t, err)
+	_, err = db.Exec(`SET CLUSTER SETTING bulkio.index_backfill.ingest_concurrency=4`)
+	require.NoError(t, err)
 	// Ensure that we checkpoint our progress to the backfill job so that
 	// RESUMEs can get an up-to-date backfill progress.
 	_, err = db.Exec(`SET CLUSTER SETTING bulkio.index_backfill.checkpoint_interval = '10ms'`)
 	require.NoError(t, err)
 	_, err = db.Exec(`CREATE TABLE t(i INT PRIMARY KEY)`)
 	require.NoError(t, err)
-	_, err = db.Exec(`INSERT INTO t SELECT generate_series(1, $1)`, numSpans)
+	// Have a 100 splits each containing a 100 rows.
+	_, err = db.Exec(`INSERT INTO t SELECT generate_series(1, $1)`, (numRows*numSpans)+1)
 	require.NoError(t, err)
-	_, err = db.Exec(`ALTER TABLE t SPLIT AT TABLE generate_series(1, $1)`, numSpans)
+	for split := 0; split < numSpans; split++ {
+		_, err = db.Exec(`ALTER TABLE t SPLIT AT VALUES ($1)`, numRows*numSpans)
+	}
 	require.NoError(t, err)
 	var descID catid.DescID
 	descIDRow := db.QueryRow(`SELECT 't'::regclass::oid`)
@@ -669,21 +685,28 @@ func TestIndexBackfillerResumePreservesProgress(t *testing.T) {
 	}
 
 	var completedSpans roachpb.SpanGroup
+	var observedSpans []roachpb.Span
 	receiveProgressUpdate := func() {
-		progressUpdate := <-backfillProgressCompletedCh
-
-		// Make sure the progress update does not contain overlapping spans.
-		for i, span1 := range progressUpdate {
-			for j, span2 := range progressUpdate {
-				if i <= j {
-					continue
-				}
-				if span1.Overlaps(span2) {
-					t.Fatalf("progress update contains overlapping spans: %s and %s", span1, span2)
+		updateCount := 2
+		for isBlockingBackfillProgress.Load() && updateCount > 0 {
+			progressUpdate := <-backfillProgressCompletedCh
+			// Make sure the progress update does not contain overlapping spans.
+			for i, span1 := range progressUpdate {
+				for j, span2 := range progressUpdate {
+					if i == j {
+						continue
+					}
+					if span1.Overlaps(span2) {
+						t.Fatalf("progress update contains overlapping spans: %s and %s", span1, span2)
+					}
 				}
 			}
+			hasMoreSpans := completedSpans.Add(progressUpdate...)
+			if hasMoreSpans {
+				updateCount -= 1
+			}
+			observedSpans = append(observedSpans, progressUpdate...)
 		}
-		completedSpans.Add(progressUpdate...)
 	}
 
 	ensureCompletedSpansAreCheckpointed := func() {
@@ -717,45 +740,35 @@ func TestIndexBackfillerResumePreservesProgress(t *testing.T) {
 		}, 5*time.Second)
 	}
 
-	// Let the backfill step forward a bit before we do our PAUSE/RESUME
-	// dance.
-	for i := 0; i < 2; i++ {
+	for isBlockingBackfillProgress.Load() {
 		receiveProgressUpdate()
-	}
-
-	ensureCompletedSpansAreCheckpointed()
-	t.Logf("pausing backfill")
-	_, err = db.Exec(`PAUSE JOB $1`, jobID)
-	require.NoError(t, err)
-	ensureJobState("paused")
-
-	t.Logf("resuming backfill")
-	_, err = db.Exec(`RESUME JOB $1`, jobID)
-	require.NoError(t, err)
-	ensureJobState("running")
+		ensureCompletedSpansAreCheckpointed()
+		t.Logf("pausing backfill")
+		_, err = db.Exec(`PAUSE JOB $1`, jobID)
+		require.NoError(t, err)
+		ensureJobState("paused")
 
-	// Step forward again before re-pausing.
-	for i := 0; i < 2; i++ {
-		receiveProgressUpdate()
+		t.Logf("resuming backfill")
+		_, err = db.Exec(`RESUME JOB $1`, jobID)
+		require.NoError(t, err)
+		ensureJobState("running")
 	}
-
-	ensureCompletedSpansAreCheckpointed()
-	isBlockingBackfillProgress.Store(false)
-
-	t.Logf("pausing backfill")
-	_, err = db.Exec(`PAUSE JOB $1`, jobID)
-	require.NoError(t, err)
-	ensureJobState("paused")
-
-	t.Logf("resuming backfill")
-	_, err = db.Exec(`RESUME JOB $1`, jobID)
-	require.NoError(t, err)
-	ensureJobState("running")
-
 	// Now we can wait for the job to succeed
 	ensureJobState("succeeded")
-
 	if err = g.Wait(); err != nil {
 		require.NoError(t, err)
 	}
+	// Make sure the spans we are adding do not overlap otherwise, this indicates
+	// a bug. Where we computed chunks incorrectly. Each chunk should be an independent
+	// piece of work.
+	for i, span1 := range observedSpans {
+		for j, span2 := range observedSpans {
+			if i == j {
+				continue
+			}
+			if span1.Overlaps(span2) {
+				t.Fatalf("progress update contains overlapping spans: %s and %s", span1, span2)
+			}
+		}
+	}
 }

pkg/sql/rowexec/indexbackfiller.go

@@ -152,9 +152,8 @@ func (ib *indexBackfiller) constructIndexEntries(
 
 			// Identify the Span for which we have constructed index entries. This is
 			// used for reporting progress and updating the job details.
-			completedSpan := ib.spec.Spans[i]
+			completedSpan := roachpb.Span{Key: startKey, EndKey: ib.spec.Spans[i].EndKey}
 			if todo.Key != nil {
-				completedSpan.Key = startKey
 				completedSpan.EndKey = todo.Key
 			}
 
@@ -510,21 +509,20 @@ func (ib *indexBackfiller) getProgressReportInterval() time.Duration {
 	return indexBackfillProgressReportInterval
 }
 
-// buildIndexEntryBatch constructs the index entries for a single indexBatch.
+// buildIndexEntryBatch constructs the index entries for a single indexBatch for
+// the given span. A non-nil resumeKey is returned when there is still work to
+// be done in this span (sp.Key < resumeKey < sp.EndKey), which happens if the
+// entire span does not fit within the batch size.
 func (ib *indexBackfiller) buildIndexEntryBatch(
 	tctx context.Context, sp roachpb.Span, readAsOf hlc.Timestamp,
-) (roachpb.Key, []rowenc.IndexEntry, int64, error) {
+) (resumeKey roachpb.Key, entries []rowenc.IndexEntry, memUsedBuildingBatch int64, err error) {
 	knobs := &ib.flowCtx.Cfg.TestingKnobs
 	if knobs.RunBeforeBackfillChunk != nil {
 		if err := knobs.RunBeforeBackfillChunk(sp); err != nil {
 			return nil, nil, 0, err
 		}
 	}
 
-	var memUsedBuildingBatch int64
-	var key roachpb.Key
-	var entries []rowenc.IndexEntry
-
 	br := indexBatchRetry{
 		nextChunkSize: ib.spec.ChunkSize,
 		// Memory used while building index entries is released by another goroutine
@@ -563,7 +561,7 @@ func (ib *indexBackfiller) buildIndexEntryBatch(
 
 		// TODO(knz): do KV tracing in DistSQL processors.
 		var err error
-		entries, key, memUsedBuildingBatch, err = ib.BuildIndexEntriesChunk(
+		entries, resumeKey, memUsedBuildingBatch, err = ib.BuildIndexEntriesChunk(
 			ctx, txn.KV(), ib.desc, sp, br.nextChunkSize, false, /* traceKV */
 		)
 		return err
@@ -579,7 +577,7 @@ func (ib *indexBackfiller) buildIndexEntryBatch(
 	log.VEventf(ctx, 3, "index backfill stats: entries %d, prepare %+v",
 		len(entries), prepTime)
 
-	return key, entries, memUsedBuildingBatch, nil
+	return resumeKey, entries, memUsedBuildingBatch, nil
 }
 
 // Resume is part of the execinfra.Processor interface.