Merge pull request #153596 from fqazi/blathers/backport-release-25.2-153583

release-25.2: sql/schemachanger: fix index creation failures after pause/resume

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.