Merge #153538 #153583 #153589

153538: roachtest: deflake disk-full roachtest r=RaduBerinde a=jbowens

This is a speculative fix for a disk-full roachtest failure. This roachtest induces an out-of-disk scenario on a node and ensures that removing the automatic emergency ballast allows recovery. In #153445 it appears disk space is exhausted on the node, but the node never notices because the write workload is insufficient to trigger any write I/O to new pages. This commit adjusts the disk-full roachtest's workload to run a little longer (10m vs 4m) and to write larger-sized values (512-byte) to ensure we continue to write a nontrivial volume of data after we intentionally exhaust available disk space.

Epic: none
Fixes: #153445.
Release note: none

153583: sql/schemachanger: fix index creation failures after pause/resume r=fqazi a=fqazi

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.

153589: kvnemesis: correctly handle errors from DeleteRangeUsingTombstoneOperation r=miraradeva a=stevendanna

Previously we were initializing the result in the
DeleteRangeUsingTombstoneOperation case using the err variable that had been set to nil.

Now, we correctly using the Err from the result. While here, I've also re-arranged the code so that the cases are a bit more consistent.

Epic: none
Release note: None

Co-authored-by: Jackson Owens <[email protected]>
Co-authored-by: Faizan Qazi <[email protected]>
Co-authored-by: Steven Danna <[email protected]>

Author: 4 people

pkg/cmd/roachtest/tests/disk_full.go

@@ -54,7 +54,8 @@ func registerDiskFull(r registry.Registry) {
 			m.Go(func(ctx context.Context) error {
 				cmd := fmt.Sprintf(
 					"./cockroach workload run kv --tolerate-errors --init --read-percent=0"+
-						" --concurrency=10 --duration=4m {pgurl:2-%d}",
+						" --min-block-bytes=512 --max-block-bytes=512"+
+						" --concurrency=10 --duration=10m {pgurl:2-%d}",
 					len(c.CRDBNodes()))
 				c.Run(ctx, option.WithNodes(c.WorkloadNode()), cmd)
 				return nil

pkg/kv/kvnemesis/applier.go

@@ -640,18 +640,17 @@ func applyBatchOp(
 			}
 			subO.Result.ResumeSpan = res.ResumeSpan
 		case *PutOperation:
-			err := b.Results[resultIdx].Err
-			subO.Result = resultInit(ctx, err)
-			subO.Result.ResumeSpan = b.Results[resultIdx].ResumeSpan
+			res := b.Results[resultIdx]
+			subO.Result = resultInit(ctx, res.Err)
+			subO.Result.ResumeSpan = res.ResumeSpan
 		case *ScanOperation:
 			res := b.Results[resultIdx]
-			kvs, err := res.Rows, res.Err
-			if err != nil {
-				subO.Result = resultInit(ctx, err)
+			if res.Err != nil {
+				subO.Result = resultInit(ctx, res.Err)
 			} else {
 				subO.Result.Type = ResultType_Values
-				subO.Result.Values = make([]KeyValue, len(kvs))
-				for j, kv := range kvs {
+				subO.Result.Values = make([]KeyValue, len(res.Rows))
+				for j, kv := range res.Rows {
 					subO.Result.Values[j] = KeyValue{
 						Key:   []byte(kv.Key),
 						Value: kv.Value.RawBytes,
@@ -660,26 +659,27 @@ func applyBatchOp(
 			}
 			subO.Result.ResumeSpan = res.ResumeSpan
 		case *DeleteOperation:
-			err := b.Results[resultIdx].Err
-			subO.Result = resultInit(ctx, err)
-			subO.Result.ResumeSpan = b.Results[resultIdx].ResumeSpan
+			res := b.Results[resultIdx]
+			subO.Result = resultInit(ctx, res.Err)
+			subO.Result.ResumeSpan = res.ResumeSpan
 		case *DeleteRangeOperation:
 			res := b.Results[resultIdx]
-			keys, err := res.Keys, res.Err
-			if err != nil {
-				subO.Result = resultInit(ctx, err)
+			if res.Err != nil {
+				subO.Result = resultInit(ctx, res.Err)
 			} else {
 				subO.Result.Type = ResultType_Keys
-				subO.Result.Keys = make([][]byte, len(keys))
-				for j, key := range keys {
+				subO.Result.Keys = make([][]byte, len(res.Keys))
+				for j, key := range res.Keys {
 					subO.Result.Keys[j] = key
 				}
 			}
 			subO.Result.ResumeSpan = res.ResumeSpan
 		case *DeleteRangeUsingTombstoneOperation:
-			subO.Result = resultInit(ctx, err)
-			subO.Result.ResumeSpan = b.Results[resultIdx].ResumeSpan
+			res := b.Results[resultIdx]
+			subO.Result = resultInit(ctx, res.Err)
+			subO.Result.ResumeSpan = res.ResumeSpan
 		case *MutateBatchHeaderOperation:
+			// NB: MutateBatchHeaderOperation cannot fail.
 			subO.Result = resultInit(ctx, nil)
 		case *AddSSTableOperation:
 			panic(errors.AssertionFailedf(`AddSSTable cannot be used in batches`))

pkg/sql/backfill/backfill.go

@@ -1010,19 +1010,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))
@@ -1031,7 +1033,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 {
@@ -1264,7 +1266,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"
@@ -582,7 +584,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)
 
@@ -611,22 +614,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`)
@@ -670,21 +686,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() {
@@ -718,45 +741,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

@@ -162,9 +162,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
 			}
 
@@ -529,21 +528,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
@@ -582,7 +580,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
@@ -598,7 +596,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.