db, manifest: clean up error cases when picking compactions

We change some of the compaction functions to not return errors and
instead log errors or panic in invariant builds.  We also add a
missing return check for `setupInputs`.

Author: RaduBerinde

compaction_picker.go

@@ -1482,6 +1482,8 @@ func (p *compactionPickerByScore) pickedCompactionFromCandidateFile(
 	}
 
 	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).
 		return nil
 	}
 
@@ -1610,6 +1612,7 @@ func pickAutoLPositive(
 	}
 
 	if !pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
+		opts.Logger.Errorf("%v", base.AssertionFailedf("setupInputs failed"))
 		return nil
 	}
 	return pc.maybeAddLevel(opts, env.diskAvailBytes)
@@ -1740,56 +1743,49 @@ func (wa WriteAmpHeuristic) String() string {
 
 // Helper method to pick compactions originating from L0. Uses information about
 // sublevels to generate a compaction.
-func pickL0(env compactionEnv, opts *Options, vers *version, baseLevel int) (pc *pickedCompaction) {
+func pickL0(env compactionEnv, opts *Options, vers *version, baseLevel int) *pickedCompaction {
 	// It is important to pass information about Lbase files to L0Sublevels
 	// so it can pick a compaction that does not conflict with an Lbase => Lbase+1
 	// compaction. Without this, we observed reduced concurrency of L0=>Lbase
 	// compactions, and increasing read amplification in L0.
 	//
 	// TODO(bilal) Remove the minCompactionDepth parameter once fixing it at 1
 	// has been shown to not cause a performance regression.
-	lcf, err := vers.L0Sublevels.PickBaseCompaction(1, vers.Levels[baseLevel].Slice())
-	if err != nil {
-		opts.Logger.Errorf("error when picking base compaction: %s", err)
-		return
-	}
+	lcf := vers.L0Sublevels.PickBaseCompaction(opts.Logger, 1, vers.Levels[baseLevel].Slice())
 	if lcf != nil {
-		pc = newPickedCompactionFromL0(lcf, opts, vers, baseLevel, true)
-		pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel)
-		if pc.startLevel.files.Empty() {
-			opts.Logger.Fatalf("empty compaction chosen")
+		pc := newPickedCompactionFromL0(lcf, opts, vers, baseLevel, true)
+		if pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
+			if pc.startLevel.files.Empty() {
+				opts.Logger.Errorf("%v", base.AssertionFailedf("empty compaction chosen"))
+			}
+			return pc.maybeAddLevel(opts, env.diskAvailBytes)
 		}
-		return pc.maybeAddLevel(opts, env.diskAvailBytes)
+		// TODO(radu): investigate why this happens.
+		// opts.Logger.Errorf("%v", base.AssertionFailedf("setupInputs failed"))
 	}
 
 	// Couldn't choose a base compaction. Try choosing an intra-L0
 	// compaction. Note that we pass in L0CompactionThreshold here as opposed to
 	// 1, since choosing a single sublevel intra-L0 compaction is
 	// counterproductive.
-	lcf, err = vers.L0Sublevels.PickIntraL0Compaction(env.earliestUnflushedSeqNum, minIntraL0Count)
-	if err != nil {
-		opts.Logger.Errorf("error when picking intra-L0 compaction: %s", err)
-		return
-	}
+	lcf = vers.L0Sublevels.PickIntraL0Compaction(env.earliestUnflushedSeqNum, minIntraL0Count)
 	if lcf != nil {
-		pc = newPickedCompactionFromL0(lcf, opts, vers, 0, false)
-		if !pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
-			return nil
-		}
-		if pc.startLevel.files.Empty() {
-			opts.Logger.Fatalf("empty compaction chosen")
-		}
-		{
-			iter := pc.startLevel.files.Iter()
-			if iter.First() == nil || iter.Next() == nil {
-				// A single-file intra-L0 compaction is unproductive.
-				return nil
+		pc := newPickedCompactionFromL0(lcf, opts, vers, 0, false)
+		if pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
+			if pc.startLevel.files.Empty() {
+				opts.Logger.Fatalf("empty compaction chosen")
+			}
+			// A single-file intra-L0 compaction is unproductive.
+			if iter := pc.startLevel.files.Iter(); iter.First() != nil && iter.Next() != nil {
+				pc.smallest, pc.largest = manifest.KeyRange(pc.cmp, pc.startLevel.files.Iter())
+				return pc
 			}
+		} else {
+			// TODO(radu): investigate why this happens.
+			// opts.Logger.Errorf("%v", base.AssertionFailedf("setupInputs failed"))
 		}
-
-		pc.smallest, pc.largest = manifest.KeyRange(pc.cmp, pc.startLevel.files.Iter())
 	}
-	return pc
+	return nil
 }
 
 func newPickedManualCompaction(

internal/manifest/l0_sublevels.go

@@ -1414,8 +1414,8 @@ func (is intervalSorterByDecreasingScore) Swap(i, j int) {
 // files that can be selected for compaction. Returns nil if no compaction is
 // possible.
 func (s *L0Sublevels) PickBaseCompaction(
-	minCompactionDepth int, baseFiles LevelSlice,
-) (*L0CompactionFiles, error) {
+	logger base.Logger, minCompactionDepth int, baseFiles LevelSlice,
+) *L0CompactionFiles {
 	// For LBase compactions, we consider intervals in a greedy manner in the
 	// following order:
 	// - Intervals that are unlikely to be blocked due
@@ -1463,21 +1463,19 @@ func (s *L0Sublevels) PickBaseCompaction(
 		// file since they are likely nearby. Note that it is possible that
 		// those intervals have seed files at lower sub-levels so could be
 		// viable for compaction.
-		if f == nil {
-			return nil, errors.New("no seed file found in sublevel intervals")
-		}
 		consideredIntervals.markBits(f.minIntervalIndex, f.maxIntervalIndex+1)
 		if f.IsCompacting() {
 			if f.IsIntraL0Compacting {
 				// If we're picking a base compaction and we came across a seed
 				// file candidate that's being intra-L0 compacted, skip the
-				// interval instead of erroring out.
+				// interval instead of emitting an error.
 				continue
 			}
-			// We chose a compaction seed file that should not be compacting.
-			// Usually means the score is not accurately accounting for files
-			// already compacting, or internal state is inconsistent.
-			return nil, errors.Errorf("file %s chosen as seed file for compaction should not be compacting", f.FileNum)
+			// We chose a compaction seed file that should not be compacting; this
+			// indicates that the the internal state is inconsistent. Note that
+			// base.AssertionFailedf panics in invariant builds.
+			logger.Errorf("%v", base.AssertionFailedf("seed file %s should not be compacting", f.FileNum))
+			continue
 		}
 
 		c := s.baseCompactionUsingSeed(f, interval.index, minCompactionDepth)
@@ -1503,10 +1501,10 @@ func (s *L0Sublevels) PickBaseCompaction(
 			if baseCompacting {
 				continue
 			}
-			return c, nil
+			return c
 		}
 	}
-	return nil, nil
+	return nil
 }
 
 // Helper function for building an L0 -> Lbase compaction using a seed interval
@@ -1640,7 +1638,7 @@ func (s *L0Sublevels) extendFiles(
 // selection.
 func (s *L0Sublevels) PickIntraL0Compaction(
 	earliestUnflushedSeqNum base.SeqNum, minCompactionDepth int,
-) (*L0CompactionFiles, error) {
+) *L0CompactionFiles {
 	scoredIntervals := make([]intervalAndScore, len(s.orderedIntervals))
 	for i := range s.orderedIntervals {
 		interval := &s.orderedIntervals[i]
@@ -1661,49 +1659,44 @@ func (s *L0Sublevels) PickIntraL0Compaction(
 			continue
 		}
 
-		var f *TableMetadata
 		// Pick the seed file for the interval as the file in the highest
 		// sub-level.
-		stackDepthReduction := scoredInterval.score
-		for i := len(interval.files) - 1; i >= 0; i-- {
-			f = interval.files[i]
-			if f.IsCompacting() {
-				break
-			}
-			consideredIntervals.markBits(f.minIntervalIndex, f.maxIntervalIndex+1)
-			// Can this be the seed file? Files with newer sequence numbers than
-			// earliestUnflushedSeqNum cannot be in the compaction.
-			if f.LargestSeqNum >= earliestUnflushedSeqNum {
+		seedFile := func() *TableMetadata {
+			stackDepthReduction := scoredInterval.score
+			for i := len(interval.files) - 1; i >= 0; i-- {
+				f := interval.files[i]
+				if f.IsCompacting() {
+					// This file could be in a concurrent intra-L0 or base compaction; we
+					// can't use this interval.
+					return nil
+				}
+				consideredIntervals.markBits(f.minIntervalIndex, f.maxIntervalIndex+1)
+				// Can this be the seed file? Files with newer sequence numbers than
+				// earliestUnflushedSeqNum cannot be in the compaction.
+				if f.LargestSeqNum < earliestUnflushedSeqNum {
+					return f
+				}
 				stackDepthReduction--
-				if stackDepthReduction == 0 {
-					break
+				if stackDepthReduction < minCompactionDepth {
+					// Can't use this interval.
+					return nil
 				}
-			} else {
-				break
 			}
-		}
-		if stackDepthReduction < minCompactionDepth {
-			// Can't use this interval.
-			continue
-		}
-
-		if f == nil {
-			return nil, errors.New("no seed file found in sublevel intervals")
-		}
-		if f.IsCompacting() {
-			// This file could be in a concurrent intra-L0 or base compaction.
+			return nil
+		}()
+		if seedFile == nil {
 			// Try another interval.
 			continue
 		}
 
 		// We have a seed file. Build a compaction off of that seed.
 		c := s.intraL0CompactionUsingSeed(
-			f, interval.index, earliestUnflushedSeqNum, minCompactionDepth)
+			seedFile, interval.index, earliestUnflushedSeqNum, minCompactionDepth)
 		if c != nil {
-			return c, nil
+			return c
 		}
 	}
-	return nil, nil
+	return nil
 }
 
 func (s *L0Sublevels) intraL0CompactionUsingSeed(

internal/manifest/l0_sublevels_test.go

@@ -354,8 +354,8 @@ func TestL0Sublevels(t *testing.T) {
 			var lcf *L0CompactionFiles
 			if pickBaseCompaction {
 				baseFiles := NewLevelSliceKeySorted(base.DefaultComparer.Compare, fileMetas[baseLevel])
-				lcf, err = sublevels.PickBaseCompaction(minCompactionDepth, baseFiles)
-				if err == nil && lcf != nil {
+				lcf = sublevels.PickBaseCompaction(base.DefaultLogger, minCompactionDepth, baseFiles)
+				if lcf != nil {
 					// Try to extend the base compaction into a more rectangular
 					// shape, using the smallest/largest keys of the files before
 					// and after overlapping base files. This mimics the logic
@@ -381,10 +381,7 @@ func TestL0Sublevels(t *testing.T) {
 						lcf)
 				}
 			} else {
-				lcf, err = sublevels.PickIntraL0Compaction(earliestUnflushedSeqNum, minCompactionDepth)
-			}
-			if err != nil {
-				return fmt.Sprintf("error: %s", err.Error())
+				lcf = sublevels.PickIntraL0Compaction(earliestUnflushedSeqNum, minCompactionDepth)
 			}
 			if lcf == nil {
 				return "no compaction picked"
@@ -610,8 +607,7 @@ func BenchmarkL0SublevelsInitAndPick(b *testing.B) {
 		if sl == nil {
 			b.Fatal("expected non-nil L0Sublevels to be generated")
 		}
-		c, err := sl.PickBaseCompaction(2, LevelSlice{})
-		require.NoError(b, err)
+		c := sl.PickBaseCompaction(base.DefaultLogger, 2, LevelSlice{})
 		if c == nil {
 			b.Fatal("expected non-nil compaction to be generated")
 		}