db: check problem spans when picking auto compactions

We check the problem spans to avoid any overlap when picking
compactions. Note that normally the problem spans are empty and these
checks are avoided.

We add a test that verifies that compactions still go through after
trying to compact a non-existent external file.

Fixes #4285
Informs #1192

Author: RaduBerinde

compaction.go

@@ -1788,7 +1788,7 @@ func (d *DB) makeCompactionEnvLocked() *compactionEnv {
 	if d.closed.Load() != nil || d.opts.ReadOnly {
 		return nil
 	}
-	return &compactionEnv{
+	env := &compactionEnv{
 		diskAvailBytes:          d.diskAvailBytes.Load(),
 		earliestSnapshotSeqNum:  d.mu.snapshots.earliest(),
 		earliestUnflushedSeqNum: d.getEarliestUnflushedSeqNumLocked(),
@@ -1799,6 +1799,10 @@ func (d *DB) makeCompactionEnvLocked() *compactionEnv {
 			rescheduleReadCompaction: &d.mu.compact.rescheduleReadCompaction,
 		},
 	}
+	if !d.problemSpans.IsEmpty() {
+		env.problemSpans = &d.problemSpans
+	}
+	return env
 }
 
 // pickAnyCompaction tries to pick a manual or automatic compaction.
@@ -2324,7 +2328,7 @@ func (d *DB) compact(c *compaction, errChannel chan error) {
 		d.mu.Lock()
 		c.grantHandle.Started()
 		if err := d.compact1(c, errChannel); err != nil {
-			d.handleCompactFailure(err)
+			d.handleCompactFailure(c, err)
 		}
 		if c.isDownload {
 			d.mu.compact.downloadingCount--
@@ -2362,13 +2366,34 @@ func (d *DB) compact(c *compaction, errChannel chan error) {
 	})
 }
 
-func (d *DB) handleCompactFailure(err error) {
+func (d *DB) handleCompactFailure(c *compaction, err error) {
 	if errors.Is(err, ErrCancelledCompaction) {
 		// ErrCancelledCompaction is expected during normal operation, so we don't
 		// want to report it as a background error.
 		d.opts.Logger.Infof("%v", err)
 		return
 	}
+
+	// Record problem spans for a short duration, unless the error is a corruption.
+	expiration := 30 * time.Second
+	if IsCorruptionError(err) {
+		// TODO(radu): ideally, we should be using the corruption reporting
+		// mechanism which has a tighter span for the corruption. We would need to
+		// somehow plumb the level of the file.
+		expiration = 5 * time.Minute
+	}
+	for i := range c.inputs {
+		level := c.inputs[i].level
+		if level == 0 {
+			// We do not set problem spans on L0, as they could block flushes.
+			continue
+		}
+		it := c.inputs[i].files.Iter()
+		for f := it.First(); f != nil; f = it.Next() {
+			d.problemSpans.Add(level, f.UserKeyBounds(), expiration)
+		}
+	}
+
 	// TODO(peter): count consecutive compaction errors and backoff.
 	d.opts.EventListener.BackgroundError(err)
 }

compaction_picker.go

@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/manifest"
+	"github.com/cockroachdb/pebble/internal/problemspans"
 )
 
 // The minimum count for an intra-L0 compaction. This matches the RocksDB
@@ -39,6 +40,10 @@ type compactionEnv struct {
 	earliestSnapshotSeqNum  base.SeqNum
 	inProgressCompactions   []compactionInfo
 	readCompactionEnv       readCompactionEnv
+	// problemSpans is checked by the compaction picker to avoid compactions that
+	// overlap an active "problem span". It can be nil when there are no problem
+	// spans.
+	problemSpans *problemspans.ByLevel
 }
 
 type compactionPicker interface {
@@ -414,7 +419,10 @@ func (pc *pickedCompaction) maybeExpandBounds(smallest InternalKey, largest Inte
 // setupInputs returns true if a compaction has been set up. It returns false if
 // a concurrent compaction is occurring on the start or output level files.
 func (pc *pickedCompaction) setupInputs(
-	opts *Options, diskAvailBytes uint64, startLevel *compactionLevel,
+	opts *Options,
+	diskAvailBytes uint64,
+	startLevel *compactionLevel,
+	problemSpans *problemspans.ByLevel,
 ) bool {
 	// maxExpandedBytes is the maximum size of an expanded compaction. If
 	// growing a compaction results in a larger size, the original compaction
@@ -423,7 +431,7 @@ func (pc *pickedCompaction) setupInputs(
 		opts, adjustedOutputLevel(pc.outputLevel.level, pc.baseLevel), diskAvailBytes,
 	)
 
-	if anyTablesCompacting(startLevel.files) {
+	if !canCompactTables(startLevel.files, startLevel.level, problemSpans) {
 		return false
 	}
 
@@ -434,7 +442,7 @@ func (pc *pickedCompaction) setupInputs(
 	// left empty for those.
 	if startLevel.level != pc.outputLevel.level {
 		pc.outputLevel.files = pc.version.Overlaps(pc.outputLevel.level, pc.userKeyBounds())
-		if anyTablesCompacting(pc.outputLevel.files) {
+		if !canCompactTables(pc.outputLevel.files, pc.outputLevel.level, problemSpans) {
 			return false
 		}
 
@@ -504,7 +512,7 @@ func (pc *pickedCompaction) setupInputs(
 				*pc.lcf = *oldLcf
 			}
 		}
-	} else if pc.grow(pc.smallest, pc.largest, maxExpandedBytes, startLevel) {
+	} else if pc.grow(pc.smallest, pc.largest, maxExpandedBytes, startLevel, problemSpans) {
 		pc.maybeExpandBounds(manifest.KeyRange(pc.cmp,
 			startLevel.files.All(), pc.outputLevel.files.All()))
 	}
@@ -521,13 +529,16 @@ func (pc *pickedCompaction) setupInputs(
 // c.level+1 files in the compaction, and returns whether the inputs grew. sm
 // and la are the smallest and largest InternalKeys in all of the inputs.
 func (pc *pickedCompaction) grow(
-	sm, la InternalKey, maxExpandedBytes uint64, startLevel *compactionLevel,
+	sm, la InternalKey,
+	maxExpandedBytes uint64,
+	startLevel *compactionLevel,
+	problemSpans *problemspans.ByLevel,
 ) bool {
 	if pc.outputLevel.files.Empty() {
 		return false
 	}
 	grow0 := pc.version.Overlaps(startLevel.level, base.UserKeyBoundsFromInternal(sm, la))
-	if anyTablesCompacting(grow0) {
+	if !canCompactTables(grow0, startLevel.level, problemSpans) {
 		return false
 	}
 	if grow0.Len() <= startLevel.files.Len() {
@@ -540,10 +551,10 @@ func (pc *pickedCompaction) grow(
 	// sm1 and la1 could shift the output level keyspace when pc.outputLevel.files is set to grow1.
 	sm1, la1 := manifest.KeyRange(pc.cmp, grow0.All(), pc.outputLevel.files.All())
 	grow1 := pc.version.Overlaps(pc.outputLevel.level, base.UserKeyBoundsFromInternal(sm1, la1))
-	if anyTablesCompacting(grow1) {
+	if grow1.Len() != pc.outputLevel.files.Len() {
 		return false
 	}
-	if grow1.Len() != pc.outputLevel.files.Len() {
+	if !canCompactTables(grow1, pc.outputLevel.level, problemSpans) {
 		return false
 	}
 	startLevel.files = grow0
@@ -570,18 +581,23 @@ func (pc *pickedCompaction) setupMultiLevelCandidate(opts *Options, diskAvailByt
 	pc.startLevel = &pc.inputs[0]
 	pc.extraLevels = []*compactionLevel{&pc.inputs[1]}
 	pc.outputLevel = &pc.inputs[2]
-	return pc.setupInputs(opts, diskAvailBytes, pc.extraLevels[len(pc.extraLevels)-1])
+	return pc.setupInputs(opts, diskAvailBytes, pc.extraLevels[len(pc.extraLevels)-1], nil /* TODO(radu) */)
 }
 
-// anyTablesCompacting returns true if any tables in the level slice are
-// compacting.
-func anyTablesCompacting(inputs manifest.LevelSlice) bool {
+// canCompactTables returns true if the tables in the level slice are not
+// compacting already and don't intersect any problem spans.
+func canCompactTables(
+	inputs manifest.LevelSlice, level int, problemSpans *problemspans.ByLevel,
+) bool {
 	for f := range inputs.All() {
 		if f.IsCompacting() {
-			return true
+			return false
+		}
+		if problemSpans != nil && problemSpans.Overlaps(level, f.UserKeyBounds()) {
+			return false
 		}
 	}
-	return false
+	return true
 }
 
 // newCompactionPickerByScore creates a compactionPickerByScore associated with
@@ -1033,6 +1049,7 @@ func pickCompactionSeedFile(
 	opts *Options,
 	level, outputLevel int,
 	earliestSnapshotSeqNum base.SeqNum,
+	problemSpans *problemspans.ByLevel,
 ) (manifest.LevelFile, bool) {
 	// Select the file within the level to compact. We want to minimize write
 	// amplification, but also ensure that (a) deletes are propagated to the
@@ -1067,22 +1084,36 @@ func pickCompactionSeedFile(
 
 	for f := startIter.First(); f != nil; f = startIter.Next() {
 		var overlappingBytes uint64
-		compacting := f.IsCompacting()
-		if compacting {
+		if f.IsCompacting() {
 			// Move on if this file is already being compacted. We'll likely
 			// still need to move past the overlapping output files regardless,
 			// but in cases where all start-level files are compacting we won't.
 			continue
 		}
+		if problemSpans != nil && problemSpans.Overlaps(level, f.UserKeyBounds()) {
+			// File touches problem span which temporarily disallows auto compactions.
+			continue
+		}
 
 		// Trim any output-level files smaller than f.
 		for outputFile != nil && sstableKeyCompare(cmp, outputFile.Largest, f.Smallest) < 0 {
 			outputFile = outputIter.Next()
 		}
 
-		for outputFile != nil && sstableKeyCompare(cmp, outputFile.Smallest, f.Largest) <= 0 && !compacting {
+		skip := false
+		for outputFile != nil && sstableKeyCompare(cmp, outputFile.Smallest, f.Largest) <= 0 {
 			overlappingBytes += outputFile.Size
-			compacting = compacting || outputFile.IsCompacting()
+			if outputFile.IsCompacting() {
+				// If one of the overlapping files is compacting, we're not going to be
+				// able to compact f anyway, so skip it.
+				skip = true
+				break
+			}
+			if problemSpans != nil && problemSpans.Overlaps(outputLevel, outputFile.UserKeyBounds()) {
+				// Overlapping file touches problem span which temporarily disallows auto compactions.
+				skip = true
+				break
+			}
 
 			// For files in the bottommost level of the LSM, the
 			// Stats.RangeDeletionsBytesEstimate field is set to the estimate
@@ -1129,11 +1160,7 @@ func pickCompactionSeedFile(
 			}
 			outputFile = outputIter.Next()
 		}
-
-		// If the input level file or one of the overlapping files is
-		// compacting, we're not going to be able to compact this file
-		// anyways, so skip it.
-		if compacting {
+		if skip {
 			continue
 		}
 
@@ -1312,7 +1339,7 @@ func (p *compactionPickerByScore) pickAuto(env compactionEnv) (pc *pickedCompact
 
 		// info.level > 0
 		var ok bool
-		info.file, ok = pickCompactionSeedFile(p.vers, p.virtualBackings, p.opts, info.level, info.outputLevel, env.earliestSnapshotSeqNum)
+		info.file, ok = pickCompactionSeedFile(p.vers, p.virtualBackings, p.opts, info.level, info.outputLevel, env.earliestSnapshotSeqNum, env.problemSpans)
 		if !ok {
 			continue
 		}
@@ -1504,9 +1531,7 @@ func (p *compactionPickerByScore) pickedCompactionFromCandidateFile(
 		return nil
 	}
 
-	if !pc.setupInputs(p.opts, env.diskAvailBytes, pc.startLevel) {
-		// TODO(radu): do we expect this to happen? (it does seem to happen if I add
-		// a log here).
+	if !pc.setupInputs(p.opts, env.diskAvailBytes, pc.startLevel, env.problemSpans) {
 		return nil
 	}
 
@@ -1639,8 +1664,7 @@ func pickAutoLPositive(
 		}
 	}
 
-	if !pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
-		opts.Logger.Errorf("%v", base.AssertionFailedf("setupInputs failed"))
+	if !pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel, env.problemSpans) {
 		return nil
 	}
 	return pc.maybeAddLevel(opts, env.diskAvailBytes)
@@ -1781,10 +1805,10 @@ func pickL0(
 	//
 	// TODO(bilal) Remove the minCompactionDepth parameter once fixing it at 1
 	// has been shown to not cause a performance regression.
-	lcf := l0Organizer.PickBaseCompaction(opts.Logger, 1, vers.Levels[baseLevel].Slice())
+	lcf := l0Organizer.PickBaseCompaction(opts.Logger, 1, vers.Levels[baseLevel].Slice(), baseLevel, env.problemSpans)
 	if lcf != nil {
 		pc := newPickedCompactionFromL0(lcf, opts, vers, l0Organizer, baseLevel, true)
-		if pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
+		if pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel, env.problemSpans) {
 			if pc.startLevel.files.Empty() {
 				opts.Logger.Errorf("%v", base.AssertionFailedf("empty compaction chosen"))
 			}
@@ -1798,10 +1822,10 @@ func pickL0(
 	// compaction. Note that we pass in L0CompactionThreshold here as opposed to
 	// 1, since choosing a single sublevel intra-L0 compaction is
 	// counterproductive.
-	lcf = l0Organizer.PickIntraL0Compaction(env.earliestUnflushedSeqNum, minIntraL0Count)
+	lcf = l0Organizer.PickIntraL0Compaction(env.earliestUnflushedSeqNum, minIntraL0Count, env.problemSpans)
 	if lcf != nil {
 		pc := newPickedCompactionFromL0(lcf, opts, vers, l0Organizer, 0, false)
-		if pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
+		if pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel, env.problemSpans) {
 			if pc.startLevel.files.Empty() {
 				opts.Logger.Fatalf("empty compaction chosen")
 			}
@@ -1854,7 +1878,9 @@ func newPickedManualCompaction(
 		// Nothing to do
 		return nil, false
 	}
-	if !pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
+	// We use nil problemSpans because we don't want problem spans to prevent
+	// manual compactions.
+	if !pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel, nil /* problemSpans */) {
 		// setupInputs returned false indicating there's a conflicting
 		// concurrent compaction.
 		return nil, true
@@ -1899,7 +1925,7 @@ func pickDownloadCompaction(
 	pc = newPickedCompaction(opts, vers, l0Organizer, level, level, baseLevel)
 	pc.kind = kind
 	pc.startLevel.files = manifest.NewLevelSliceKeySorted(opts.Comparer.Compare, []*tableMetadata{file})
-	if !pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
+	if !pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel, nil /* problemSpans */) {
 		// setupInputs returned false indicating there's a conflicting
 		// concurrent compaction.
 		return nil
@@ -1950,7 +1976,7 @@ func pickReadTriggeredCompactionHelper(
 	pc = newPickedCompaction(p.opts, p.vers, p.l0Organizer, rc.level, defaultOutputLevel(rc.level, p.baseLevel), p.baseLevel)
 
 	pc.startLevel.files = overlapSlice
-	if !pc.setupInputs(p.opts, env.diskAvailBytes, pc.startLevel) {
+	if !pc.setupInputs(p.opts, env.diskAvailBytes, pc.startLevel, env.problemSpans) {
 		return nil
 	}
 	if inputRangeAlreadyCompacting(env, pc) {

compaction_picker_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/manifest"
+	"github.com/cockroachdb/pebble/internal/problemspans"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
@@ -1108,7 +1109,7 @@ func TestPickedCompactionSetupInputs(t *testing.T) {
 			)
 
 			var isCompacting bool
-			if !pc.setupInputs(opts, availBytes, pc.startLevel) {
+			if !pc.setupInputs(opts, availBytes, pc.startLevel, nil /* problemSpans */) {
 				isCompacting = true
 			}
 			origPC := pc
@@ -1410,9 +1411,12 @@ func TestCompactionPickerPickFile(t *testing.T) {
 		}
 	}()
 
+	var problemSpans *problemspans.ByLevel
+
 	datadriven.RunTest(t, "testdata/compaction_picker_pick_file", func(t *testing.T, td *datadriven.TestData) string {
 		switch td.Cmd {
 		case "define":
+			problemSpans = nil
 			require.NoError(t, d.Close())
 
 			d, err = runDBDefineCmd(td, opts)
@@ -1445,6 +1449,21 @@ func TestCompactionPickerPickFile(t *testing.T) {
 			d.mu.Unlock()
 			return s
 
+		case "problem-spans":
+			problemSpans = &problemspans.ByLevel{}
+			problemSpans.Init(manifest.NumLevels, opts.Comparer.Compare)
+			for _, line := range crstrings.Lines(td.Input) {
+				var level int
+				var span1, span2 string
+				n, err := fmt.Sscanf(line, "L%d %s %s", &level, &span1, &span2)
+				if err != nil || n != 3 {
+					td.Fatalf(t, "malformed problem span %q", line)
+				}
+				bounds := base.ParseUserKeyBounds(span1 + " " + span2)
+				problemSpans.Add(level, bounds, time.Hour*10)
+			}
+			return ""
+
 		case "pick-file":
 			s := strings.TrimPrefix(td.CmdArgs[0].String(), "L")
 			level, err := strconv.Atoi(s)
@@ -1471,7 +1490,7 @@ func TestCompactionPickerPickFile(t *testing.T) {
 					return
 				}
 				p := d.mu.versions.picker.(*compactionPickerByScore)
-				lf, ok = pickCompactionSeedFile(p.vers, p.virtualBackings, opts, level, level+1, env.earliestSnapshotSeqNum)
+				lf, ok = pickCompactionSeedFile(p.vers, p.virtualBackings, opts, level, level+1, env.earliestSnapshotSeqNum, problemSpans)
 			}()
 			if !ok {
 				return "(none)"