db: deflake TestCompactionCorruption

This change attempts to deflake this test by removing unrelated
external files and allowing the test to control the range of keys in
the workload.

Fixes #5039

Author: RaduBerinde

compaction_test.go

@@ -3156,15 +3156,13 @@ func TestCompactionCorruption(t *testing.T) {
 	now.Store(1)
 	d.problemSpans.InitForTesting(manifest.NumLevels, d.cmp, func() crtime.Mono { return now.Load() })
 
-	var workloadWG sync.WaitGroup
-	var stopWorkload atomic.Bool
-	defer stopWorkload.Store(true)
-	startWorkload := func() {
-		stopWorkload.Store(false)
-		workloadWG.Add(1)
+	startWorkload := func(minKey, maxKey byte) (stop func()) {
+		var shouldStop atomic.Bool
+		var wg sync.WaitGroup
+		wg.Add(1)
 		go func() {
-			defer workloadWG.Done()
-			for !stopWorkload.Load() {
+			defer wg.Done()
+			for !shouldStop.Load() {
 				b := d.NewBatch()
 				// Write a random key of the form a012345 and flush it. This will result
 				// in (mostly) non-overlapping tables in L0.
@@ -3174,7 +3172,7 @@ func TestCompactionCorruption(t *testing.T) {
 				}
 				v := make([]byte, 1024+rand.IntN(10240))
 				_, _ = rand.NewChaCha8(valSeed).Read(v)
-				key := fmt.Sprintf("%c%06d", 'a'+byte(rand.IntN(int('z'-'a'+1))), rand.IntN(1000000))
+				key := fmt.Sprintf("%c%06d", minKey+byte(rand.IntN(int(maxKey-minKey+1))), rand.IntN(1000000))
 				if err := b.Set([]byte(key), v, nil); err != nil {
 					panic(err)
 				}
@@ -3187,9 +3185,19 @@ func TestCompactionCorruption(t *testing.T) {
 				time.Sleep(10 * time.Millisecond)
 			}
 		}()
+		return func() {
+			shouldStop.Store(true)
+			wg.Wait()
+		}
 	}
 
 	datadriven.RunTest(t, "testdata/compaction_corruption", func(t *testing.T, td *datadriven.TestData) string {
+		if arg, ok := td.Arg("workload"); ok {
+			if len(arg.Vals) != 2 || len(arg.Vals[0]) != 1 || len(arg.Vals[1]) != 1 {
+				td.Fatalf(t, "workload argument must be of the form (a,z)")
+			}
+			defer startWorkload(arg.Vals[0][0], arg.Vals[1][0])()
+		}
 		// wait until fn() returns true.
 		wait := func(what string, fn func() bool) {
 			const timeout = 2 * time.Minute
@@ -3230,13 +3238,6 @@ func TestCompactionCorruption(t *testing.T) {
 			require.NoError(t, writer.Close())
 			return fmt.Sprintf("%s -> %s", before, after)
 
-		case "start-workload":
-			startWorkload()
-
-		case "stop-workload":
-			stopWorkload.Store(true)
-			workloadWG.Wait()
-
 		case "wait-for-problem-span":
 			wait("problem span", func() bool {
 				return !d.problemSpans.IsEmpty()

testdata/compaction_corruption

@@ -1,73 +1,53 @@
-build-remote file1
-a#0,SET = avalue
-b#0,SET = bvalue
-c#0,SET = cvalue
-----
-
-build-remote file2-not-there
+build-remote file-not-there
 d#0,SET = dvalue
 q#0,SET = qvalue
 w#0,SET = wvalue
 ----
 
-build-remote file3
-x#0,SET = xvalue
-y#0,SET = yvalue
-z#0,SET = zvalue
-----
-
 ingest-external
-file1 bounds=(a,c0)
-file2 bounds=(d,w0)
-file3 bounds=(x,z0)
-----
-
-start-workload
+file bounds=(d,w0)
 ----
 
 # Verify that a problem span is set.
-wait-for-problem-span
+wait-for-problem-span workload=(d,w)
 ----
 
 # Verify that compactions still go through.
-wait-for-compactions
+wait-for-compactions workload=(a,z)
 ----
 
-# Make file2 appear.
-move-remote-object file2-not-there file2
+# Make file appear.
+move-remote-object file-not-there file
 ----
-file2-not-there -> file2
+file-not-there -> file
 
 # Expire spans.
 expire-spans
 ----
 
 # Compactions should now go through and eventually there should be no external
 # files.
-wait-for-no-external-files
+wait-for-no-external-files workload=(d,w)
 ----
 
-build-remote file4-not-there
+build-remote file2-not-there
 a#0,SET = avalue
 u#0,SET = uvalue
 z#0,SET = zvalue
 ----
 
 ingest-external
-file4 bounds=(a,z0)
+file2 bounds=(a,z0)
 ----
 
 # Verify that a problem span is set.
-wait-for-problem-span
+wait-for-problem-span workload=(a,z)
 ----
 
-stop-workload
-----
-
-# Make file4 appear.
-move-remote-object file4-not-there file4
+# Make file2 appear.
+move-remote-object file2-not-there file2
 ----
-file4-not-there -> file4
+file2-not-there -> file2
 
 # Verify that a manual compaction goes through despite the problem span.
 manual-compaction