LCOV - code coverage report
Current view: top level - pebble - compaction_picker.go (source / functions) Hit Total Coverage
Test: 2024-10-09 08:16Z e1b330bd - meta test only.lcov Lines: 993 1196 83.0 %
Date: 2024-10-09 08:17:21 Functions: 0 0 -

          Line data    Source code
       1             : // Copyright 2018 The LevelDB-Go and Pebble Authors. All rights reserved. Use
       2             : // of this source code is governed by a BSD-style license that can be found in
       3             : // the LICENSE file.
       4             : 
       5             : package pebble
       6             : 
       7             : import (
       8             :         "bytes"
       9             :         "fmt"
      10             :         "math"
      11             :         "sort"
      12             :         "strings"
      13             : 
      14             :         "github.com/cockroachdb/errors"
      15             :         "github.com/cockroachdb/pebble/internal/base"
      16             :         "github.com/cockroachdb/pebble/internal/humanize"
      17             :         "github.com/cockroachdb/pebble/internal/invariants"
      18             :         "github.com/cockroachdb/pebble/internal/manifest"
      19             : )
      20             : 
      21             : // The minimum count for an intra-L0 compaction. This matches the RocksDB
      22             : // heuristic.
      23             : const minIntraL0Count = 4
      24             : 
      25             : type compactionEnv struct {
      26             :         // diskAvailBytes holds a statistic on the number of bytes available on
      27             :         // disk, as reported by the filesystem. It's used to be more restrictive in
      28             :         // expanding compactions if available disk space is limited.
      29             :         //
      30             :         // The cached value (d.diskAvailBytes) is updated whenever a file is deleted
      31             :         // and whenever a compaction or flush completes. Since file removal is the
      32             :         // primary means of reclaiming space, there is a rough bound on the
      33             :         // statistic's staleness when available bytes is growing. Compactions and
      34             :         // flushes are longer, slower operations and provide a much looser bound
      35             :         // when available bytes is decreasing.
      36             :         diskAvailBytes          uint64
      37             :         earliestUnflushedSeqNum base.SeqNum
      38             :         earliestSnapshotSeqNum  base.SeqNum
      39             :         inProgressCompactions   []compactionInfo
      40             :         readCompactionEnv       readCompactionEnv
      41             : }
      42             : 
      43             : type compactionPicker interface {
      44             :         getScores([]compactionInfo) [numLevels]float64
      45             :         getBaseLevel() int
      46             :         estimatedCompactionDebt(l0ExtraSize uint64) uint64
      47             :         pickAuto(env compactionEnv) (pc *pickedCompaction)
      48             :         pickElisionOnlyCompaction(env compactionEnv) (pc *pickedCompaction)
      49             :         pickRewriteCompaction(env compactionEnv) (pc *pickedCompaction)
      50             :         pickReadTriggeredCompaction(env compactionEnv) (pc *pickedCompaction)
      51             :         forceBaseLevel1()
      52             : }
      53             : 
      54             : // readCompactionEnv is used to hold data required to perform read compactions
      55             : type readCompactionEnv struct {
      56             :         rescheduleReadCompaction *bool
      57             :         readCompactions          *readCompactionQueue
      58             :         flushing                 bool
      59             : }
      60             : 
      61             : // Information about in-progress compactions provided to the compaction picker.
      62             : // These are used to constrain the new compactions that will be picked.
      63             : type compactionInfo struct {
      64             :         // versionEditApplied is true if this compaction's version edit has already
      65             :         // been committed. The compaction may still be in-progress deleting newly
      66             :         // obsolete files.
      67             :         versionEditApplied bool
      68             :         inputs             []compactionLevel
      69             :         outputLevel        int
      70             :         smallest           InternalKey
      71             :         largest            InternalKey
      72             : }
      73             : 
      74           0 : func (info compactionInfo) String() string {
      75           0 :         var buf bytes.Buffer
      76           0 :         var largest int
      77           0 :         for i, in := range info.inputs {
      78           0 :                 if i > 0 {
      79           0 :                         fmt.Fprintf(&buf, " -> ")
      80           0 :                 }
      81           0 :                 fmt.Fprintf(&buf, "L%d", in.level)
      82           0 :                 in.files.Each(func(m *fileMetadata) {
      83           0 :                         fmt.Fprintf(&buf, " %s", m.FileNum)
      84           0 :                 })
      85           0 :                 if largest < in.level {
      86           0 :                         largest = in.level
      87           0 :                 }
      88             :         }
      89           0 :         if largest != info.outputLevel || len(info.inputs) == 1 {
      90           0 :                 fmt.Fprintf(&buf, " -> L%d", info.outputLevel)
      91           0 :         }
      92           0 :         return buf.String()
      93             : }
      94             : 
      95             : type sortCompactionLevelsByPriority []candidateLevelInfo
      96             : 
      97           1 : func (s sortCompactionLevelsByPriority) Len() int {
      98           1 :         return len(s)
      99           1 : }
     100             : 
     101             : // A level should be picked for compaction if the compensatedScoreRatio is >= the
     102             : // compactionScoreThreshold.
     103             : const compactionScoreThreshold = 1
     104             : 
     105             : // Less should return true if s[i] must be placed earlier than s[j] in the final
     106             : // sorted list. The candidateLevelInfo for the level placed earlier is more likely
     107             : // to be picked for a compaction.
     108           1 : func (s sortCompactionLevelsByPriority) Less(i, j int) bool {
     109           1 :         iShouldCompact := s[i].compensatedScoreRatio >= compactionScoreThreshold
     110           1 :         jShouldCompact := s[j].compensatedScoreRatio >= compactionScoreThreshold
     111           1 :         // Ordering is defined as decreasing on (shouldCompact, uncompensatedScoreRatio)
     112           1 :         // where shouldCompact is 1 for true and 0 for false.
     113           1 :         if iShouldCompact && !jShouldCompact {
     114           1 :                 return true
     115           1 :         }
     116           1 :         if !iShouldCompact && jShouldCompact {
     117           1 :                 return false
     118           1 :         }
     119             : 
     120           1 :         if s[i].uncompensatedScoreRatio != s[j].uncompensatedScoreRatio {
     121           1 :                 return s[i].uncompensatedScoreRatio > s[j].uncompensatedScoreRatio
     122           1 :         }
     123           1 :         return s[i].level < s[j].level
     124             : }
     125             : 
     126           1 : func (s sortCompactionLevelsByPriority) Swap(i, j int) {
     127           1 :         s[i], s[j] = s[j], s[i]
     128           1 : }
     129             : 
     130             : // sublevelInfo is used to tag a LevelSlice for an L0 sublevel with the
     131             : // sublevel.
     132             : type sublevelInfo struct {
     133             :         manifest.LevelSlice
     134             :         sublevel manifest.Layer
     135             : }
     136             : 
     137           1 : func (cl sublevelInfo) Clone() sublevelInfo {
     138           1 :         return sublevelInfo{
     139           1 :                 sublevel:   cl.sublevel,
     140           1 :                 LevelSlice: cl.LevelSlice,
     141           1 :         }
     142           1 : }
     143           0 : func (cl sublevelInfo) String() string {
     144           0 :         return fmt.Sprintf(`Sublevel %s; Levels %s`, cl.sublevel, cl.LevelSlice)
     145           0 : }
     146             : 
     147             : // generateSublevelInfo will generate the level slices for each of the sublevels
     148             : // from the level slice for all of L0.
     149           1 : func generateSublevelInfo(cmp base.Compare, levelFiles manifest.LevelSlice) []sublevelInfo {
     150           1 :         sublevelMap := make(map[uint64][]*fileMetadata)
     151           1 :         it := levelFiles.Iter()
     152           1 :         for f := it.First(); f != nil; f = it.Next() {
     153           1 :                 sublevelMap[uint64(f.SubLevel)] = append(sublevelMap[uint64(f.SubLevel)], f)
     154           1 :         }
     155             : 
     156           1 :         var sublevels []int
     157           1 :         for level := range sublevelMap {
     158           1 :                 sublevels = append(sublevels, int(level))
     159           1 :         }
     160           1 :         sort.Ints(sublevels)
     161           1 : 
     162           1 :         var levelSlices []sublevelInfo
     163           1 :         for _, sublevel := range sublevels {
     164           1 :                 metas := sublevelMap[uint64(sublevel)]
     165           1 :                 levelSlices = append(
     166           1 :                         levelSlices,
     167           1 :                         sublevelInfo{
     168           1 :                                 manifest.NewLevelSliceKeySorted(cmp, metas),
     169           1 :                                 manifest.L0Sublevel(sublevel),
     170           1 :                         },
     171           1 :                 )
     172           1 :         }
     173           1 :         return levelSlices
     174             : }
     175             : 
     176             : // compactionPickerMetrics holds metrics related to the compaction picking process
     177             : type compactionPickerMetrics struct {
     178             :         // scores contains the compensatedScoreRatio from the candidateLevelInfo.
     179             :         scores                      []float64
     180             :         singleLevelOverlappingRatio float64
     181             :         multiLevelOverlappingRatio  float64
     182             : }
     183             : 
     184             : // pickedCompaction contains information about a compaction that has already
     185             : // been chosen, and is being constructed. Compaction construction info lives in
     186             : // this struct, and is copied over into the compaction struct when that's
     187             : // created.
     188             : type pickedCompaction struct {
     189             :         cmp Compare
     190             :         // score of the chosen compaction. This is the same as the
     191             :         // compensatedScoreRatio in the candidateLevelInfo.
     192             :         score float64
     193             :         // kind indicates the kind of compaction.
     194             :         kind compactionKind
     195             :         // startLevel is the level that is being compacted. Inputs from startLevel
     196             :         // and outputLevel will be merged to produce a set of outputLevel files.
     197             :         startLevel *compactionLevel
     198             :         // outputLevel is the level that files are being produced in. outputLevel is
     199             :         // equal to startLevel+1 except when:
     200             :         //    - if startLevel is 0, the output level equals compactionPicker.baseLevel().
     201             :         //    - in multilevel compaction, the output level is the lowest level involved in
     202             :         //      the compaction
     203             :         outputLevel *compactionLevel
     204             :         // extraLevels contain additional levels in between the input and output
     205             :         // levels that get compacted in multi level compactions
     206             :         extraLevels []*compactionLevel
     207             :         inputs      []compactionLevel
     208             :         // LBase at the time of compaction picking.
     209             :         baseLevel int
     210             :         // L0-specific compaction info. Set to a non-nil value for all compactions
     211             :         // where startLevel == 0 that were generated by L0Sublevels.
     212             :         lcf *manifest.L0CompactionFiles
     213             :         // maxOutputFileSize is the maximum size of an individual table created
     214             :         // during compaction.
     215             :         maxOutputFileSize uint64
     216             :         // maxOverlapBytes is the maximum number of bytes of overlap allowed for a
     217             :         // single output table with the tables in the grandparent level.
     218             :         maxOverlapBytes uint64
     219             :         // maxReadCompactionBytes is the maximum bytes a read compaction is allowed to
     220             :         // overlap in its output level with. If the overlap is greater than
     221             :         // maxReadCompaction bytes, then we don't proceed with the compaction.
     222             :         maxReadCompactionBytes uint64
     223             :         // The boundaries of the input data.
     224             :         smallest      InternalKey
     225             :         largest       InternalKey
     226             :         version       *version
     227             :         pickerMetrics compactionPickerMetrics
     228             : }
     229             : 
     230           1 : func (pc *pickedCompaction) userKeyBounds() base.UserKeyBounds {
     231           1 :         return base.UserKeyBoundsFromInternal(pc.smallest, pc.largest)
     232           1 : }
     233             : 
     234           1 : func defaultOutputLevel(startLevel, baseLevel int) int {
     235           1 :         outputLevel := startLevel + 1
     236           1 :         if startLevel == 0 {
     237           1 :                 outputLevel = baseLevel
     238           1 :         }
     239           1 :         if outputLevel >= numLevels-1 {
     240           1 :                 outputLevel = numLevels - 1
     241           1 :         }
     242           1 :         return outputLevel
     243             : }
     244             : 
     245             : func newPickedCompaction(
     246             :         opts *Options, cur *version, startLevel, outputLevel, baseLevel int,
     247           1 : ) *pickedCompaction {
     248           1 :         if startLevel > 0 && startLevel < baseLevel {
     249           0 :                 panic(fmt.Sprintf("invalid compaction: start level %d should not be empty (base level %d)",
     250           0 :                         startLevel, baseLevel))
     251             :         }
     252             : 
     253           1 :         adjustedLevel := adjustedOutputLevel(outputLevel, baseLevel)
     254           1 :         pc := &pickedCompaction{
     255           1 :                 cmp:                    opts.Comparer.Compare,
     256           1 :                 version:                cur,
     257           1 :                 baseLevel:              baseLevel,
     258           1 :                 inputs:                 []compactionLevel{{level: startLevel}, {level: outputLevel}},
     259           1 :                 maxOutputFileSize:      uint64(opts.Level(adjustedLevel).TargetFileSize),
     260           1 :                 maxOverlapBytes:        maxGrandparentOverlapBytes(opts, adjustedLevel),
     261           1 :                 maxReadCompactionBytes: maxReadCompactionBytes(opts, adjustedLevel),
     262           1 :         }
     263           1 :         pc.startLevel = &pc.inputs[0]
     264           1 :         pc.outputLevel = &pc.inputs[1]
     265           1 :         return pc
     266             : }
     267             : 
     268             : // adjustedOutputLevel is the output level used for the purpose of
     269             : // determining the target output file size, overlap bytes, and expanded
     270             : // bytes, taking into account the base level.
     271           1 : func adjustedOutputLevel(outputLevel int, baseLevel int) int {
     272           1 :         adjustedOutputLevel := outputLevel
     273           1 :         if adjustedOutputLevel > 0 {
     274           1 :                 // Output level is in the range [baseLevel, numLevels]. For the purpose of
     275           1 :                 // determining the target output file size, overlap bytes, and expanded
     276           1 :                 // bytes, we want to adjust the range to [1,numLevels].
     277           1 :                 adjustedOutputLevel = 1 + outputLevel - baseLevel
     278           1 :         }
     279           1 :         return adjustedOutputLevel
     280             : }
     281             : 
     282             : func newPickedCompactionFromL0(
     283             :         lcf *manifest.L0CompactionFiles, opts *Options, vers *version, baseLevel int, isBase bool,
     284           1 : ) *pickedCompaction {
     285           1 :         outputLevel := baseLevel
     286           1 :         if !isBase {
     287           1 :                 outputLevel = 0 // Intra L0
     288           1 :         }
     289             : 
     290           1 :         pc := newPickedCompaction(opts, vers, 0, outputLevel, baseLevel)
     291           1 :         pc.lcf = lcf
     292           1 :         pc.outputLevel.level = outputLevel
     293           1 : 
     294           1 :         // Manually build the compaction as opposed to calling
     295           1 :         // pickAutoHelper. This is because L0Sublevels has already added
     296           1 :         // any overlapping L0 SSTables that need to be added, and
     297           1 :         // because compactions built by L0SSTables do not necessarily
     298           1 :         // pick contiguous sequences of files in pc.version.Levels[0].
     299           1 :         files := make([]*manifest.FileMetadata, 0, len(lcf.Files))
     300           1 :         iter := vers.Levels[0].Iter()
     301           1 :         for f := iter.First(); f != nil; f = iter.Next() {
     302           1 :                 if lcf.FilesIncluded[f.L0Index] {
     303           1 :                         files = append(files, f)
     304           1 :                 }
     305             :         }
     306           1 :         pc.startLevel.files = manifest.NewLevelSliceSeqSorted(files)
     307           1 :         return pc
     308             : }
     309             : 
     310           0 : func (pc *pickedCompaction) String() string {
     311           0 :         var builder strings.Builder
     312           0 :         builder.WriteString(fmt.Sprintf(`Score=%f, `, pc.score))
     313           0 :         builder.WriteString(fmt.Sprintf(`Kind=%s, `, pc.kind))
     314           0 :         builder.WriteString(fmt.Sprintf(`AdjustedOutputLevel=%d, `, adjustedOutputLevel(pc.outputLevel.level, pc.baseLevel)))
     315           0 :         builder.WriteString(fmt.Sprintf(`maxOutputFileSize=%d, `, pc.maxOutputFileSize))
     316           0 :         builder.WriteString(fmt.Sprintf(`maxReadCompactionBytes=%d, `, pc.maxReadCompactionBytes))
     317           0 :         builder.WriteString(fmt.Sprintf(`smallest=%s, `, pc.smallest))
     318           0 :         builder.WriteString(fmt.Sprintf(`largest=%s, `, pc.largest))
     319           0 :         builder.WriteString(fmt.Sprintf(`version=%s, `, pc.version))
     320           0 :         builder.WriteString(fmt.Sprintf(`inputs=%s, `, pc.inputs))
     321           0 :         builder.WriteString(fmt.Sprintf(`startlevel=%s, `, pc.startLevel))
     322           0 :         builder.WriteString(fmt.Sprintf(`outputLevel=%s, `, pc.outputLevel))
     323           0 :         builder.WriteString(fmt.Sprintf(`extraLevels=%s, `, pc.extraLevels))
     324           0 :         builder.WriteString(fmt.Sprintf(`l0SublevelInfo=%s, `, pc.startLevel.l0SublevelInfo))
     325           0 :         builder.WriteString(fmt.Sprintf(`lcf=%s`, pc.lcf))
     326           0 :         return builder.String()
     327           0 : }
     328             : 
     329             : // Clone creates a deep copy of the pickedCompaction
     330           1 : func (pc *pickedCompaction) clone() *pickedCompaction {
     331           1 : 
     332           1 :         // Quickly copy over fields that do not require special deep copy care, and
     333           1 :         // set all fields that will require a deep copy to nil.
     334           1 :         newPC := &pickedCompaction{
     335           1 :                 cmp:                    pc.cmp,
     336           1 :                 score:                  pc.score,
     337           1 :                 kind:                   pc.kind,
     338           1 :                 baseLevel:              pc.baseLevel,
     339           1 :                 maxOutputFileSize:      pc.maxOutputFileSize,
     340           1 :                 maxOverlapBytes:        pc.maxOverlapBytes,
     341           1 :                 maxReadCompactionBytes: pc.maxReadCompactionBytes,
     342           1 :                 smallest:               pc.smallest.Clone(),
     343           1 :                 largest:                pc.largest.Clone(),
     344           1 : 
     345           1 :                 // TODO(msbutler): properly clone picker metrics
     346           1 :                 pickerMetrics: pc.pickerMetrics,
     347           1 : 
     348           1 :                 // Both copies see the same manifest, therefore, it's ok for them to se
     349           1 :                 // share the same pc. version.
     350           1 :                 version: pc.version,
     351           1 :         }
     352           1 : 
     353           1 :         newPC.inputs = make([]compactionLevel, len(pc.inputs))
     354           1 :         newPC.extraLevels = make([]*compactionLevel, 0, len(pc.extraLevels))
     355           1 :         for i := range pc.inputs {
     356           1 :                 newPC.inputs[i] = pc.inputs[i].Clone()
     357           1 :                 if i == 0 {
     358           1 :                         newPC.startLevel = &newPC.inputs[i]
     359           1 :                 } else if i == len(pc.inputs)-1 {
     360           1 :                         newPC.outputLevel = &newPC.inputs[i]
     361           1 :                 } else {
     362           0 :                         newPC.extraLevels = append(newPC.extraLevels, &newPC.inputs[i])
     363           0 :                 }
     364             :         }
     365             : 
     366           1 :         if len(pc.startLevel.l0SublevelInfo) > 0 {
     367           1 :                 newPC.startLevel.l0SublevelInfo = make([]sublevelInfo, len(pc.startLevel.l0SublevelInfo))
     368           1 :                 for i := range pc.startLevel.l0SublevelInfo {
     369           1 :                         newPC.startLevel.l0SublevelInfo[i] = pc.startLevel.l0SublevelInfo[i].Clone()
     370           1 :                 }
     371             :         }
     372           1 :         if pc.lcf != nil {
     373           1 :                 newPC.lcf = pc.lcf.Clone()
     374           1 :         }
     375           1 :         return newPC
     376             : }
     377             : 
     378             : // maybeExpandBounds is a helper function for setupInputs which ensures the
     379             : // pickedCompaction's smallest and largest internal keys are updated iff
     380             : // the candidate keys expand the key span. This avoids a bug for multi-level
     381             : // compactions: during the second call to setupInputs, the picked compaction's
     382             : // smallest and largest keys should not decrease the key span.
     383           1 : func (pc *pickedCompaction) maybeExpandBounds(smallest InternalKey, largest InternalKey) {
     384           1 :         if len(smallest.UserKey) == 0 && len(largest.UserKey) == 0 {
     385           1 :                 return
     386           1 :         }
     387           1 :         if len(pc.smallest.UserKey) == 0 && len(pc.largest.UserKey) == 0 {
     388           1 :                 pc.smallest = smallest
     389           1 :                 pc.largest = largest
     390           1 :                 return
     391           1 :         }
     392           1 :         if base.InternalCompare(pc.cmp, pc.smallest, smallest) >= 0 {
     393           1 :                 pc.smallest = smallest
     394           1 :         }
     395           1 :         if base.InternalCompare(pc.cmp, pc.largest, largest) <= 0 {
     396           1 :                 pc.largest = largest
     397           1 :         }
     398             : }
     399             : 
     400             : // setupInputs returns true if a compaction has been set up. It returns false if
     401             : // a concurrent compaction is occurring on the start or output level files.
     402             : func (pc *pickedCompaction) setupInputs(
     403             :         opts *Options, diskAvailBytes uint64, startLevel *compactionLevel,
     404           1 : ) bool {
     405           1 :         // maxExpandedBytes is the maximum size of an expanded compaction. If
     406           1 :         // growing a compaction results in a larger size, the original compaction
     407           1 :         // is used instead.
     408           1 :         maxExpandedBytes := expandedCompactionByteSizeLimit(
     409           1 :                 opts, adjustedOutputLevel(pc.outputLevel.level, pc.baseLevel), diskAvailBytes,
     410           1 :         )
     411           1 : 
     412           1 :         if anyTablesCompacting(startLevel.files) {
     413           1 :                 return false
     414           1 :         }
     415             : 
     416           1 :         pc.maybeExpandBounds(manifest.KeyRange(pc.cmp, startLevel.files.Iter()))
     417           1 : 
     418           1 :         // Determine the sstables in the output level which overlap with the input
     419           1 :         // sstables. No need to do this for intra-L0 compactions; outputLevel.files is
     420           1 :         // left empty for those.
     421           1 :         if startLevel.level != pc.outputLevel.level {
     422           1 :                 pc.outputLevel.files = pc.version.Overlaps(pc.outputLevel.level, pc.userKeyBounds())
     423           1 :                 if anyTablesCompacting(pc.outputLevel.files) {
     424           1 :                         return false
     425           1 :                 }
     426             : 
     427           1 :                 pc.maybeExpandBounds(manifest.KeyRange(pc.cmp,
     428           1 :                         startLevel.files.Iter(), pc.outputLevel.files.Iter()))
     429             :         }
     430             : 
     431             :         // Grow the sstables in startLevel.level as long as it doesn't affect the number
     432             :         // of sstables included from pc.outputLevel.level.
     433           1 :         if pc.lcf != nil && startLevel.level == 0 && pc.outputLevel.level != 0 {
     434           1 :                 // Call the L0-specific compaction extension method. Similar logic as
     435           1 :                 // pc.grow. Additional L0 files are optionally added to the compaction at
     436           1 :                 // this step. Note that the bounds passed in are not the bounds of the
     437           1 :                 // compaction, but rather the smallest and largest internal keys that
     438           1 :                 // the compaction cannot include from L0 without pulling in more Lbase
     439           1 :                 // files. Consider this example:
     440           1 :                 //
     441           1 :                 // L0:        c-d e+f g-h
     442           1 :                 // Lbase: a-b     e+f     i-j
     443           1 :                 //        a b c d e f g h i j
     444           1 :                 //
     445           1 :                 // The e-f files have already been chosen in the compaction. As pulling
     446           1 :                 // in more LBase files is undesirable, the logic below will pass in
     447           1 :                 // smallest = b and largest = i to ExtendL0ForBaseCompactionTo, which
     448           1 :                 // will expand the compaction to include c-d and g-h from L0. The
     449           1 :                 // bounds passed in are exclusive; the compaction cannot be expanded
     450           1 :                 // to include files that "touch" it.
     451           1 :                 smallestBaseKey := base.InvalidInternalKey
     452           1 :                 largestBaseKey := base.InvalidInternalKey
     453           1 :                 if pc.outputLevel.files.Empty() {
     454           1 :                         baseIter := pc.version.Levels[pc.outputLevel.level].Iter()
     455           1 :                         if sm := baseIter.SeekLT(pc.cmp, pc.smallest.UserKey); sm != nil {
     456           1 :                                 smallestBaseKey = sm.Largest
     457           1 :                         }
     458           1 :                         if la := baseIter.SeekGE(pc.cmp, pc.largest.UserKey); la != nil {
     459           1 :                                 largestBaseKey = la.Smallest
     460           1 :                         }
     461           1 :                 } else {
     462           1 :                         // NB: We use Reslice to access the underlying level's files, but
     463           1 :                         // we discard the returned slice. The pc.outputLevel.files slice
     464           1 :                         // is not modified.
     465           1 :                         _ = pc.outputLevel.files.Reslice(func(start, end *manifest.LevelIterator) {
     466           1 :                                 if sm := start.Prev(); sm != nil {
     467           1 :                                         smallestBaseKey = sm.Largest
     468           1 :                                 }
     469           1 :                                 if la := end.Next(); la != nil {
     470           1 :                                         largestBaseKey = la.Smallest
     471           1 :                                 }
     472             :                         })
     473             :                 }
     474           1 :                 oldLcf := pc.lcf.Clone()
     475           1 :                 if pc.version.L0Sublevels.ExtendL0ForBaseCompactionTo(smallestBaseKey, largestBaseKey, pc.lcf) {
     476           1 :                         var newStartLevelFiles []*fileMetadata
     477           1 :                         iter := pc.version.Levels[0].Iter()
     478           1 :                         var sizeSum uint64
     479           1 :                         for j, f := 0, iter.First(); f != nil; j, f = j+1, iter.Next() {
     480           1 :                                 if pc.lcf.FilesIncluded[f.L0Index] {
     481           1 :                                         newStartLevelFiles = append(newStartLevelFiles, f)
     482           1 :                                         sizeSum += f.Size
     483           1 :                                 }
     484             :                         }
     485           1 :                         if sizeSum+pc.outputLevel.files.SizeSum() < maxExpandedBytes {
     486           1 :                                 startLevel.files = manifest.NewLevelSliceSeqSorted(newStartLevelFiles)
     487           1 :                                 pc.smallest, pc.largest = manifest.KeyRange(pc.cmp,
     488           1 :                                         startLevel.files.Iter(), pc.outputLevel.files.Iter())
     489           1 :                         } else {
     490           1 :                                 *pc.lcf = *oldLcf
     491           1 :                         }
     492             :                 }
     493           1 :         } else if pc.grow(pc.smallest, pc.largest, maxExpandedBytes, startLevel) {
     494           1 :                 pc.maybeExpandBounds(manifest.KeyRange(pc.cmp,
     495           1 :                         startLevel.files.Iter(), pc.outputLevel.files.Iter()))
     496           1 :         }
     497             : 
     498           1 :         if pc.startLevel.level == 0 {
     499           1 :                 // We don't change the input files for the compaction beyond this point.
     500           1 :                 pc.startLevel.l0SublevelInfo = generateSublevelInfo(pc.cmp, pc.startLevel.files)
     501           1 :         }
     502             : 
     503           1 :         return true
     504             : }
     505             : 
     506             : // grow grows the number of inputs at c.level without changing the number of
     507             : // c.level+1 files in the compaction, and returns whether the inputs grew. sm
     508             : // and la are the smallest and largest InternalKeys in all of the inputs.
     509             : func (pc *pickedCompaction) grow(
     510             :         sm, la InternalKey, maxExpandedBytes uint64, startLevel *compactionLevel,
     511           1 : ) bool {
     512           1 :         if pc.outputLevel.files.Empty() {
     513           1 :                 return false
     514           1 :         }
     515           1 :         grow0 := pc.version.Overlaps(startLevel.level, base.UserKeyBoundsFromInternal(sm, la))
     516           1 :         if anyTablesCompacting(grow0) {
     517           1 :                 return false
     518           1 :         }
     519           1 :         if grow0.Len() <= startLevel.files.Len() {
     520           1 :                 return false
     521           1 :         }
     522           1 :         if grow0.SizeSum()+pc.outputLevel.files.SizeSum() >= maxExpandedBytes {
     523           1 :                 return false
     524           1 :         }
     525             :         // We need to include the outputLevel iter because without it, in a multiLevel scenario,
     526             :         // sm1 and la1 could shift the output level keyspace when pc.outputLevel.files is set to grow1.
     527           1 :         sm1, la1 := manifest.KeyRange(pc.cmp, grow0.Iter(), pc.outputLevel.files.Iter())
     528           1 :         grow1 := pc.version.Overlaps(pc.outputLevel.level, base.UserKeyBoundsFromInternal(sm1, la1))
     529           1 :         if anyTablesCompacting(grow1) {
     530           1 :                 return false
     531           1 :         }
     532           1 :         if grow1.Len() != pc.outputLevel.files.Len() {
     533           1 :                 return false
     534           1 :         }
     535           1 :         startLevel.files = grow0
     536           1 :         pc.outputLevel.files = grow1
     537           1 :         return true
     538             : }
     539             : 
     540           1 : func (pc *pickedCompaction) compactionSize() uint64 {
     541           1 :         var bytesToCompact uint64
     542           1 :         for i := range pc.inputs {
     543           1 :                 bytesToCompact += pc.inputs[i].files.SizeSum()
     544           1 :         }
     545           1 :         return bytesToCompact
     546             : }
     547             : 
     548             : // setupMultiLevelCandidated returns true if it successfully added another level
     549             : // to the compaction.
     550           1 : func (pc *pickedCompaction) setupMultiLevelCandidate(opts *Options, diskAvailBytes uint64) bool {
     551           1 :         pc.inputs = append(pc.inputs, compactionLevel{level: pc.outputLevel.level + 1})
     552           1 : 
     553           1 :         // Recalibrate startLevel and outputLevel:
     554           1 :         //  - startLevel and outputLevel pointers may be obsolete after appending to pc.inputs.
     555           1 :         //  - push outputLevel to extraLevels and move the new level to outputLevel
     556           1 :         pc.startLevel = &pc.inputs[0]
     557           1 :         pc.extraLevels = []*compactionLevel{&pc.inputs[1]}
     558           1 :         pc.outputLevel = &pc.inputs[2]
     559           1 :         return pc.setupInputs(opts, diskAvailBytes, pc.extraLevels[len(pc.extraLevels)-1])
     560           1 : }
     561             : 
     562             : // anyTablesCompacting returns true if any tables in the level slice are
     563             : // compacting.
     564           1 : func anyTablesCompacting(inputs manifest.LevelSlice) bool {
     565           1 :         it := inputs.Iter()
     566           1 :         for f := it.First(); f != nil; f = it.Next() {
     567           1 :                 if f.IsCompacting() {
     568           1 :                         return true
     569           1 :                 }
     570             :         }
     571           1 :         return false
     572             : }
     573             : 
     574             : // newCompactionPickerByScore creates a compactionPickerByScore associated with
     575             : // the newest version. The picker is used under logLock (until a new version is
     576             : // installed).
     577             : func newCompactionPickerByScore(
     578             :         v *version,
     579             :         virtualBackings *manifest.VirtualBackings,
     580             :         opts *Options,
     581             :         inProgressCompactions []compactionInfo,
     582           1 : ) *compactionPickerByScore {
     583           1 :         p := &compactionPickerByScore{
     584           1 :                 opts:            opts,
     585           1 :                 vers:            v,
     586           1 :                 virtualBackings: virtualBackings,
     587           1 :         }
     588           1 :         p.initLevelMaxBytes(inProgressCompactions)
     589           1 :         p.initTombstoneDensityAnnotator(opts)
     590           1 :         return p
     591           1 : }
     592             : 
     593             : // Information about a candidate compaction level that has been identified by
     594             : // the compaction picker.
     595             : type candidateLevelInfo struct {
     596             :         // The compensatedScore of the level after adjusting according to the other
     597             :         // levels' sizes. For L0, the compensatedScoreRatio is equivalent to the
     598             :         // uncompensatedScoreRatio as we don't account for level size compensation in
     599             :         // L0.
     600             :         compensatedScoreRatio float64
     601             :         // The score of the level after accounting for level size compensation before
     602             :         // adjusting according to other levels' sizes. For L0, the compensatedScore
     603             :         // is equivalent to the uncompensatedScore as we don't account for level
     604             :         // size compensation in L0.
     605             :         compensatedScore float64
     606             :         // The score of the level to be compacted, calculated using uncompensated file
     607             :         // sizes and without any adjustments.
     608             :         uncompensatedScore float64
     609             :         // uncompensatedScoreRatio is the uncompensatedScore adjusted according to
     610             :         // the other levels' sizes.
     611             :         uncompensatedScoreRatio float64
     612             :         level                   int
     613             :         // The level to compact to.
     614             :         outputLevel int
     615             :         // The file in level that will be compacted. Additional files may be
     616             :         // picked by the compaction, and a pickedCompaction created for the
     617             :         // compaction.
     618             :         file manifest.LevelFile
     619             : }
     620             : 
     621           1 : func (c *candidateLevelInfo) shouldCompact() bool {
     622           1 :         return c.compensatedScoreRatio >= compactionScoreThreshold
     623           1 : }
     624             : 
     625           1 : func fileCompensation(f *fileMetadata) uint64 {
     626           1 :         return uint64(f.Stats.PointDeletionsBytesEstimate) + f.Stats.RangeDeletionsBytesEstimate
     627           1 : }
     628             : 
     629             : // compensatedSize returns f's file size, inflated according to compaction
     630             : // priorities.
     631           1 : func compensatedSize(f *fileMetadata) uint64 {
     632           1 :         // Add in the estimate of disk space that may be reclaimed by compacting the
     633           1 :         // file's tombstones.
     634           1 :         return f.Size + fileCompensation(f)
     635           1 : }
     636             : 
     637             : // compensatedSizeAnnotator is a manifest.Annotator that annotates B-Tree
     638             : // nodes with the sum of the files' compensated sizes. Compensated sizes may
     639             : // change once a table's stats are loaded asynchronously, so its values are
     640             : // marked as cacheable only if a file's stats have been loaded.
     641           1 : var compensatedSizeAnnotator = manifest.SumAnnotator(func(f *fileMetadata) (uint64, bool) {
     642           1 :         return compensatedSize(f), f.StatsValid()
     643           1 : })
     644             : 
     645             : // totalCompensatedSize computes the compensated size over a file metadata
     646             : // iterator. Note that this function is linear in the files available to the
     647             : // iterator. Use the compensatedSizeAnnotator if querying the total
     648             : // compensated size of a level.
     649           1 : func totalCompensatedSize(iter manifest.LevelIterator) uint64 {
     650           1 :         var sz uint64
     651           1 :         for f := iter.First(); f != nil; f = iter.Next() {
     652           1 :                 sz += compensatedSize(f)
     653           1 :         }
     654           1 :         return sz
     655             : }
     656             : 
     657             : // compactionPickerByScore holds the state and logic for picking a compaction. A
     658             : // compaction picker is associated with a single version. A new compaction
     659             : // picker is created and initialized every time a new version is installed.
     660             : type compactionPickerByScore struct {
     661             :         opts            *Options
     662             :         vers            *version
     663             :         virtualBackings *manifest.VirtualBackings
     664             :         // The level to target for L0 compactions. Levels L1 to baseLevel must be
     665             :         // empty.
     666             :         baseLevel int
     667             :         // levelMaxBytes holds the dynamically adjusted max bytes setting for each
     668             :         // level.
     669             :         levelMaxBytes [numLevels]int64
     670             :         // tombstoneDensityAnnotator holds the annotator for choosing tombstone
     671             :         // density compactions.
     672             :         // NB: This is declared here rather than globally because
     673             :         // options.Experimental.MinTombstoneDenseRatio is not known until runtime.
     674             :         tombstoneDensityAnnotator *manifest.Annotator[fileMetadata]
     675             : }
     676             : 
     677             : var _ compactionPicker = &compactionPickerByScore{}
     678             : 
     679           1 : func (p *compactionPickerByScore) getScores(inProgress []compactionInfo) [numLevels]float64 {
     680           1 :         var scores [numLevels]float64
     681           1 :         for _, info := range p.calculateLevelScores(inProgress) {
     682           1 :                 scores[info.level] = info.compensatedScoreRatio
     683           1 :         }
     684           1 :         return scores
     685             : }
     686             : 
     687           1 : func (p *compactionPickerByScore) getBaseLevel() int {
     688           1 :         if p == nil {
     689           0 :                 return 1
     690           0 :         }
     691           1 :         return p.baseLevel
     692             : }
     693             : 
     694             : // estimatedCompactionDebt estimates the number of bytes which need to be
     695             : // compacted before the LSM tree becomes stable.
     696           1 : func (p *compactionPickerByScore) estimatedCompactionDebt(l0ExtraSize uint64) uint64 {
     697           1 :         if p == nil {
     698           0 :                 return 0
     699           0 :         }
     700             : 
     701             :         // We assume that all the bytes in L0 need to be compacted to Lbase. This is
     702             :         // unlike the RocksDB logic that figures out whether L0 needs compaction.
     703           1 :         bytesAddedToNextLevel := l0ExtraSize + p.vers.Levels[0].Size()
     704           1 :         lbaseSize := p.vers.Levels[p.baseLevel].Size()
     705           1 : 
     706           1 :         var compactionDebt uint64
     707           1 :         if bytesAddedToNextLevel > 0 && lbaseSize > 0 {
     708           1 :                 // We only incur compaction debt if both L0 and Lbase contain data. If L0
     709           1 :                 // is empty, no compaction is necessary. If Lbase is empty, a move-based
     710           1 :                 // compaction from L0 would occur.
     711           1 :                 compactionDebt += bytesAddedToNextLevel + lbaseSize
     712           1 :         }
     713             : 
     714             :         // loop invariant: At the beginning of the loop, bytesAddedToNextLevel is the
     715             :         // bytes added to `level` in the loop.
     716           1 :         for level := p.baseLevel; level < numLevels-1; level++ {
     717           1 :                 levelSize := p.vers.Levels[level].Size() + bytesAddedToNextLevel
     718           1 :                 nextLevelSize := p.vers.Levels[level+1].Size()
     719           1 :                 if levelSize > uint64(p.levelMaxBytes[level]) {
     720           1 :                         bytesAddedToNextLevel = levelSize - uint64(p.levelMaxBytes[level])
     721           1 :                         if nextLevelSize > 0 {
     722           1 :                                 // We only incur compaction debt if the next level contains data. If the
     723           1 :                                 // next level is empty, a move-based compaction would be used.
     724           1 :                                 levelRatio := float64(nextLevelSize) / float64(levelSize)
     725           1 :                                 // The current level contributes bytesAddedToNextLevel to compactions.
     726           1 :                                 // The next level contributes levelRatio * bytesAddedToNextLevel.
     727           1 :                                 compactionDebt += uint64(float64(bytesAddedToNextLevel) * (levelRatio + 1))
     728           1 :                         }
     729           1 :                 } else {
     730           1 :                         // We're not moving any bytes to the next level.
     731           1 :                         bytesAddedToNextLevel = 0
     732           1 :                 }
     733             :         }
     734           1 :         return compactionDebt
     735             : }
     736             : 
     737           1 : func (p *compactionPickerByScore) initLevelMaxBytes(inProgressCompactions []compactionInfo) {
     738           1 :         // The levelMaxBytes calculations here differ from RocksDB in two ways:
     739           1 :         //
     740           1 :         // 1. The use of dbSize vs maxLevelSize. RocksDB uses the size of the maximum
     741           1 :         //    level in L1-L6, rather than determining the size of the bottom level
     742           1 :         //    based on the total amount of data in the dB. The RocksDB calculation is
     743           1 :         //    problematic if L0 contains a significant fraction of data, or if the
     744           1 :         //    level sizes are roughly equal and thus there is a significant fraction
     745           1 :         //    of data outside of the largest level.
     746           1 :         //
     747           1 :         // 2. Not adjusting the size of Lbase based on L0. RocksDB computes
     748           1 :         //    baseBytesMax as the maximum of the configured LBaseMaxBytes and the
     749           1 :         //    size of L0. This is problematic because baseBytesMax is used to compute
     750           1 :         //    the max size of lower levels. A very large baseBytesMax will result in
     751           1 :         //    an overly large value for the size of lower levels which will caused
     752           1 :         //    those levels not to be compacted even when they should be
     753           1 :         //    compacted. This often results in "inverted" LSM shapes where Ln is
     754           1 :         //    larger than Ln+1.
     755           1 : 
     756           1 :         // Determine the first non-empty level and the total DB size.
     757           1 :         firstNonEmptyLevel := -1
     758           1 :         var dbSize uint64
     759           1 :         for level := 1; level < numLevels; level++ {
     760           1 :                 if p.vers.Levels[level].Size() > 0 {
     761           1 :                         if firstNonEmptyLevel == -1 {
     762           1 :                                 firstNonEmptyLevel = level
     763           1 :                         }
     764           1 :                         dbSize += p.vers.Levels[level].Size()
     765             :                 }
     766             :         }
     767           1 :         for _, c := range inProgressCompactions {
     768           1 :                 if c.outputLevel == 0 || c.outputLevel == -1 {
     769           1 :                         continue
     770             :                 }
     771           1 :                 if c.inputs[0].level == 0 && (firstNonEmptyLevel == -1 || c.outputLevel < firstNonEmptyLevel) {
     772           1 :                         firstNonEmptyLevel = c.outputLevel
     773           1 :                 }
     774             :         }
     775             : 
     776             :         // Initialize the max-bytes setting for each level to "infinity" which will
     777             :         // disallow compaction for that level. We'll fill in the actual value below
     778             :         // for levels we want to allow compactions from.
     779           1 :         for level := 0; level < numLevels; level++ {
     780           1 :                 p.levelMaxBytes[level] = math.MaxInt64
     781           1 :         }
     782             : 
     783           1 :         if dbSize == 0 {
     784           1 :                 // No levels for L1 and up contain any data. Target L0 compactions for the
     785           1 :                 // last level or to the level to which there is an ongoing L0 compaction.
     786           1 :                 p.baseLevel = numLevels - 1
     787           1 :                 if firstNonEmptyLevel >= 0 {
     788           1 :                         p.baseLevel = firstNonEmptyLevel
     789           1 :                 }
     790           1 :                 return
     791             :         }
     792             : 
     793           1 :         dbSize += p.vers.Levels[0].Size()
     794           1 :         bottomLevelSize := dbSize - dbSize/uint64(p.opts.Experimental.LevelMultiplier)
     795           1 : 
     796           1 :         curLevelSize := bottomLevelSize
     797           1 :         for level := numLevels - 2; level >= firstNonEmptyLevel; level-- {
     798           1 :                 curLevelSize = uint64(float64(curLevelSize) / float64(p.opts.Experimental.LevelMultiplier))
     799           1 :         }
     800             : 
     801             :         // Compute base level (where L0 data is compacted to).
     802           1 :         baseBytesMax := uint64(p.opts.LBaseMaxBytes)
     803           1 :         p.baseLevel = firstNonEmptyLevel
     804           1 :         for p.baseLevel > 1 && curLevelSize > baseBytesMax {
     805           1 :                 p.baseLevel--
     806           1 :                 curLevelSize = uint64(float64(curLevelSize) / float64(p.opts.Experimental.LevelMultiplier))
     807           1 :         }
     808             : 
     809           1 :         smoothedLevelMultiplier := 1.0
     810           1 :         if p.baseLevel < numLevels-1 {
     811           1 :                 smoothedLevelMultiplier = math.Pow(
     812           1 :                         float64(bottomLevelSize)/float64(baseBytesMax),
     813           1 :                         1.0/float64(numLevels-p.baseLevel-1))
     814           1 :         }
     815             : 
     816           1 :         levelSize := float64(baseBytesMax)
     817           1 :         for level := p.baseLevel; level < numLevels; level++ {
     818           1 :                 if level > p.baseLevel && levelSize > 0 {
     819           1 :                         levelSize *= smoothedLevelMultiplier
     820           1 :                 }
     821             :                 // Round the result since test cases use small target level sizes, which
     822             :                 // can be impacted by floating-point imprecision + integer truncation.
     823           1 :                 roundedLevelSize := math.Round(levelSize)
     824           1 :                 if roundedLevelSize > float64(math.MaxInt64) {
     825           0 :                         p.levelMaxBytes[level] = math.MaxInt64
     826           1 :                 } else {
     827           1 :                         p.levelMaxBytes[level] = int64(roundedLevelSize)
     828           1 :                 }
     829             :         }
     830             : }
     831             : 
     832             : type levelSizeAdjust struct {
     833             :         incomingActualBytes      uint64
     834             :         outgoingActualBytes      uint64
     835             :         outgoingCompensatedBytes uint64
     836             : }
     837             : 
     838           1 : func (a levelSizeAdjust) compensated() uint64 {
     839           1 :         return a.incomingActualBytes - a.outgoingCompensatedBytes
     840           1 : }
     841             : 
     842           1 : func (a levelSizeAdjust) actual() uint64 {
     843           1 :         return a.incomingActualBytes - a.outgoingActualBytes
     844           1 : }
     845             : 
     846           1 : func calculateSizeAdjust(inProgressCompactions []compactionInfo) [numLevels]levelSizeAdjust {
     847           1 :         // Compute size adjustments for each level based on the in-progress
     848           1 :         // compactions. We sum the file sizes of all files leaving and entering each
     849           1 :         // level in in-progress compactions. For outgoing files, we also sum a
     850           1 :         // separate sum of 'compensated file sizes', which are inflated according
     851           1 :         // to deletion estimates.
     852           1 :         //
     853           1 :         // When we adjust a level's size according to these values during score
     854           1 :         // calculation, we subtract the compensated size of start level inputs to
     855           1 :         // account for the fact that score calculation uses compensated sizes.
     856           1 :         //
     857           1 :         // Since compensated file sizes may be compensated because they reclaim
     858           1 :         // space from the output level's files, we only add the real file size to
     859           1 :         // the output level.
     860           1 :         //
     861           1 :         // This is slightly different from RocksDB's behavior, which simply elides
     862           1 :         // compacting files from the level size calculation.
     863           1 :         var sizeAdjust [numLevels]levelSizeAdjust
     864           1 :         for i := range inProgressCompactions {
     865           1 :                 c := &inProgressCompactions[i]
     866           1 :                 // If this compaction's version edit has already been applied, there's
     867           1 :                 // no need to adjust: The LSM we'll examine will already reflect the
     868           1 :                 // new LSM state.
     869           1 :                 if c.versionEditApplied {
     870           1 :                         continue
     871             :                 }
     872             : 
     873           1 :                 for _, input := range c.inputs {
     874           1 :                         actualSize := input.files.SizeSum()
     875           1 :                         compensatedSize := totalCompensatedSize(input.files.Iter())
     876           1 : 
     877           1 :                         if input.level != c.outputLevel {
     878           1 :                                 sizeAdjust[input.level].outgoingCompensatedBytes += compensatedSize
     879           1 :                                 sizeAdjust[input.level].outgoingActualBytes += actualSize
     880           1 :                                 if c.outputLevel != -1 {
     881           1 :                                         sizeAdjust[c.outputLevel].incomingActualBytes += actualSize
     882           1 :                                 }
     883             :                         }
     884             :                 }
     885             :         }
     886           1 :         return sizeAdjust
     887             : }
     888             : 
     889             : func (p *compactionPickerByScore) calculateLevelScores(
     890             :         inProgressCompactions []compactionInfo,
     891           1 : ) [numLevels]candidateLevelInfo {
     892           1 :         var scores [numLevels]candidateLevelInfo
     893           1 :         for i := range scores {
     894           1 :                 scores[i].level = i
     895           1 :                 scores[i].outputLevel = i + 1
     896           1 :         }
     897           1 :         l0UncompensatedScore := calculateL0UncompensatedScore(p.vers, p.opts, inProgressCompactions)
     898           1 :         scores[0] = candidateLevelInfo{
     899           1 :                 outputLevel:        p.baseLevel,
     900           1 :                 uncompensatedScore: l0UncompensatedScore,
     901           1 :                 compensatedScore:   l0UncompensatedScore, /* No level size compensation for L0 */
     902           1 :         }
     903           1 :         sizeAdjust := calculateSizeAdjust(inProgressCompactions)
     904           1 :         for level := 1; level < numLevels; level++ {
     905           1 :                 compensatedLevelSize := *compensatedSizeAnnotator.LevelAnnotation(p.vers.Levels[level]) + sizeAdjust[level].compensated()
     906           1 :                 scores[level].compensatedScore = float64(compensatedLevelSize) / float64(p.levelMaxBytes[level])
     907           1 :                 scores[level].uncompensatedScore = float64(p.vers.Levels[level].Size()+sizeAdjust[level].actual()) / float64(p.levelMaxBytes[level])
     908           1 :         }
     909             : 
     910             :         // Adjust each level's {compensated, uncompensated}Score by the uncompensatedScore
     911             :         // of the next level to get a {compensated, uncompensated}ScoreRatio. If the
     912             :         // next level has a high uncompensatedScore, and is thus a priority for compaction,
     913             :         // this reduces the priority for compacting the current level. If the next level
     914             :         // has a low uncompensatedScore (i.e. it is below its target size), this increases
     915             :         // the priority for compacting the current level.
     916             :         //
     917             :         // The effect of this adjustment is to help prioritize compactions in lower
     918             :         // levels. The following example shows the compensatedScoreRatio and the
     919             :         // compensatedScore. In this scenario, L0 has 68 sublevels. L3 (a.k.a. Lbase)
     920             :         // is significantly above its target size. The original score prioritizes
     921             :         // compactions from those two levels, but doing so ends up causing a future
     922             :         // problem: data piles up in the higher levels, starving L5->L6 compactions,
     923             :         // and to a lesser degree starving L4->L5 compactions.
     924             :         //
     925             :         // Note that in the example shown there is no level size compensation so the
     926             :         // compensatedScore and the uncompensatedScore is the same for each level.
     927             :         //
     928             :         //        compensatedScoreRatio   compensatedScore   uncompensatedScore   size   max-size
     929             :         //   L0                     3.2               68.0                 68.0  2.2 G          -
     930             :         //   L3                     3.2               21.1                 21.1  1.3 G       64 M
     931             :         //   L4                     3.4                6.7                  6.7  3.1 G      467 M
     932             :         //   L5                     3.4                2.0                  2.0  6.6 G      3.3 G
     933             :         //   L6                     0.6                0.6                  0.6   14 G       24 G
     934           1 :         var prevLevel int
     935           1 :         for level := p.baseLevel; level < numLevels; level++ {
     936           1 :                 // The compensated scores, and uncompensated scores will be turned into
     937           1 :                 // ratios as they're adjusted according to other levels' sizes.
     938           1 :                 scores[prevLevel].compensatedScoreRatio = scores[prevLevel].compensatedScore
     939           1 :                 scores[prevLevel].uncompensatedScoreRatio = scores[prevLevel].uncompensatedScore
     940           1 : 
     941           1 :                 // Avoid absurdly large scores by placing a floor on the score that we'll
     942           1 :                 // adjust a level by. The value of 0.01 was chosen somewhat arbitrarily.
     943           1 :                 const minScore = 0.01
     944           1 :                 if scores[prevLevel].compensatedScoreRatio >= compactionScoreThreshold {
     945           1 :                         if scores[level].uncompensatedScore >= minScore {
     946           1 :                                 scores[prevLevel].compensatedScoreRatio /= scores[level].uncompensatedScore
     947           1 :                         } else {
     948           1 :                                 scores[prevLevel].compensatedScoreRatio /= minScore
     949           1 :                         }
     950             :                 }
     951           1 :                 if scores[prevLevel].uncompensatedScoreRatio >= compactionScoreThreshold {
     952           1 :                         if scores[level].uncompensatedScore >= minScore {
     953           1 :                                 scores[prevLevel].uncompensatedScoreRatio /= scores[level].uncompensatedScore
     954           1 :                         } else {
     955           1 :                                 scores[prevLevel].uncompensatedScoreRatio /= minScore
     956           1 :                         }
     957             :                 }
     958           1 :                 prevLevel = level
     959             :         }
     960             :         // Set the score ratios for the lowest level.
     961             :         // INVARIANT: prevLevel == numLevels-1
     962           1 :         scores[prevLevel].compensatedScoreRatio = scores[prevLevel].compensatedScore
     963           1 :         scores[prevLevel].uncompensatedScoreRatio = scores[prevLevel].uncompensatedScore
     964           1 : 
     965           1 :         sort.Sort(sortCompactionLevelsByPriority(scores[:]))
     966           1 :         return scores
     967             : }
     968             : 
     969             : // calculateL0UncompensatedScore calculates a float score representing the
     970             : // relative priority of compacting L0. Level L0 is special in that files within
     971             : // L0 may overlap one another, so a different set of heuristics that take into
     972             : // account read amplification apply.
     973             : func calculateL0UncompensatedScore(
     974             :         vers *version, opts *Options, inProgressCompactions []compactionInfo,
     975           1 : ) float64 {
     976           1 :         // Use the sublevel count to calculate the score. The base vs intra-L0
     977           1 :         // compaction determination happens in pickAuto, not here.
     978           1 :         score := float64(2*vers.L0Sublevels.MaxDepthAfterOngoingCompactions()) /
     979           1 :                 float64(opts.L0CompactionThreshold)
     980           1 : 
     981           1 :         // Also calculate a score based on the file count but use it only if it
     982           1 :         // produces a higher score than the sublevel-based one. This heuristic is
     983           1 :         // designed to accommodate cases where L0 is accumulating non-overlapping
     984           1 :         // files in L0. Letting too many non-overlapping files accumulate in few
     985           1 :         // sublevels is undesirable, because:
     986           1 :         // 1) we can produce a massive backlog to compact once files do overlap.
     987           1 :         // 2) constructing L0 sublevels has a runtime that grows superlinearly with
     988           1 :         //    the number of files in L0 and must be done while holding D.mu.
     989           1 :         noncompactingFiles := vers.Levels[0].Len()
     990           1 :         for _, c := range inProgressCompactions {
     991           1 :                 for _, cl := range c.inputs {
     992           1 :                         if cl.level == 0 {
     993           1 :                                 noncompactingFiles -= cl.files.Len()
     994           1 :                         }
     995             :                 }
     996             :         }
     997           1 :         fileScore := float64(noncompactingFiles) / float64(opts.L0CompactionFileThreshold)
     998           1 :         if score < fileScore {
     999           1 :                 score = fileScore
    1000           1 :         }
    1001           1 :         return score
    1002             : }
    1003             : 
    1004             : // pickCompactionSeedFile picks a file from `level` in the `vers` to build a
    1005             : // compaction around. Currently, this function implements a heuristic similar to
    1006             : // RocksDB's kMinOverlappingRatio, seeking to minimize write amplification. This
    1007             : // function is linear with respect to the number of files in `level` and
    1008             : // `outputLevel`.
    1009             : func pickCompactionSeedFile(
    1010             :         vers *version,
    1011             :         virtualBackings *manifest.VirtualBackings,
    1012             :         opts *Options,
    1013             :         level, outputLevel int,
    1014             :         earliestSnapshotSeqNum base.SeqNum,
    1015           1 : ) (manifest.LevelFile, bool) {
    1016           1 :         // Select the file within the level to compact. We want to minimize write
    1017           1 :         // amplification, but also ensure that (a) deletes are propagated to the
    1018           1 :         // bottom level in a timely fashion, and (b) virtual sstables that are
    1019           1 :         // pinning backing sstables where most of the data is garbage are compacted
    1020           1 :         // away. Doing (a) and (b) reclaims disk space. A table's smallest sequence
    1021           1 :         // number provides a measure of its age. The ratio of overlapping-bytes /
    1022           1 :         // table-size gives an indication of write amplification (a smaller ratio is
    1023           1 :         // preferrable).
    1024           1 :         //
    1025           1 :         // The current heuristic is based off the RocksDB kMinOverlappingRatio
    1026           1 :         // heuristic. It chooses the file with the minimum overlapping ratio with
    1027           1 :         // the target level, which minimizes write amplification.
    1028           1 :         //
    1029           1 :         // The heuristic uses a "compensated size" for the denominator, which is the
    1030           1 :         // file size inflated by (a) an estimate of the space that may be reclaimed
    1031           1 :         // through compaction, and (b) a fraction of the amount of garbage in the
    1032           1 :         // backing sstable pinned by this (virtual) sstable.
    1033           1 :         //
    1034           1 :         // TODO(peter): For concurrent compactions, we may want to try harder to
    1035           1 :         // pick a seed file whose resulting compaction bounds do not overlap with
    1036           1 :         // an in-progress compaction.
    1037           1 : 
    1038           1 :         cmp := opts.Comparer.Compare
    1039           1 :         startIter := vers.Levels[level].Iter()
    1040           1 :         outputIter := vers.Levels[outputLevel].Iter()
    1041           1 : 
    1042           1 :         var file manifest.LevelFile
    1043           1 :         smallestRatio := uint64(math.MaxUint64)
    1044           1 : 
    1045           1 :         outputFile := outputIter.First()
    1046           1 : 
    1047           1 :         for f := startIter.First(); f != nil; f = startIter.Next() {
    1048           1 :                 var overlappingBytes uint64
    1049           1 :                 compacting := f.IsCompacting()
    1050           1 :                 if compacting {
    1051           1 :                         // Move on if this file is already being compacted. We'll likely
    1052           1 :                         // still need to move past the overlapping output files regardless,
    1053           1 :                         // but in cases where all start-level files are compacting we won't.
    1054           1 :                         continue
    1055             :                 }
    1056             : 
    1057             :                 // Trim any output-level files smaller than f.
    1058           1 :                 for outputFile != nil && sstableKeyCompare(cmp, outputFile.Largest, f.Smallest) < 0 {
    1059           1 :                         outputFile = outputIter.Next()
    1060           1 :                 }
    1061             : 
    1062           1 :                 for outputFile != nil && sstableKeyCompare(cmp, outputFile.Smallest, f.Largest) <= 0 && !compacting {
    1063           1 :                         overlappingBytes += outputFile.Size
    1064           1 :                         compacting = compacting || outputFile.IsCompacting()
    1065           1 : 
    1066           1 :                         // For files in the bottommost level of the LSM, the
    1067           1 :                         // Stats.RangeDeletionsBytesEstimate field is set to the estimate
    1068           1 :                         // of bytes /within/ the file itself that may be dropped by
    1069           1 :                         // recompacting the file. These bytes from obsolete keys would not
    1070           1 :                         // need to be rewritten if we compacted `f` into `outputFile`, so
    1071           1 :                         // they don't contribute to write amplification. Subtracting them
    1072           1 :                         // out of the overlapping bytes helps prioritize these compactions
    1073           1 :                         // that are cheaper than their file sizes suggest.
    1074           1 :                         if outputLevel == numLevels-1 && outputFile.LargestSeqNum < earliestSnapshotSeqNum {
    1075           1 :                                 overlappingBytes -= outputFile.Stats.RangeDeletionsBytesEstimate
    1076           1 :                         }
    1077             : 
    1078             :                         // If the file in the next level extends beyond f's largest key,
    1079             :                         // break out and don't advance outputIter because f's successor
    1080             :                         // might also overlap.
    1081             :                         //
    1082             :                         // Note, we stop as soon as we encounter an output-level file with a
    1083             :                         // largest key beyond the input-level file's largest bound. We
    1084             :                         // perform a simple user key comparison here using sstableKeyCompare
    1085             :                         // which handles the potential for exclusive largest key bounds.
    1086             :                         // There's some subtlety when the bounds are equal (eg, equal and
    1087             :                         // inclusive, or equal and exclusive). Current Pebble doesn't split
    1088             :                         // user keys across sstables within a level (and in format versions
    1089             :                         // FormatSplitUserKeysMarkedCompacted and later we guarantee no
    1090             :                         // split user keys exist within the entire LSM). In that case, we're
    1091             :                         // assured that neither the input level nor the output level's next
    1092             :                         // file shares the same user key, so compaction expansion will not
    1093             :                         // include them in any compaction compacting `f`.
    1094             :                         //
    1095             :                         // NB: If we /did/ allow split user keys, or we're running on an
    1096             :                         // old database with an earlier format major version where there are
    1097             :                         // existing split user keys, this logic would be incorrect. Consider
    1098             :                         //    L1: [a#120,a#100] [a#80,a#60]
    1099             :                         //    L2: [a#55,a#45] [a#35,a#25] [a#15,a#5]
    1100             :                         // While considering the first file in L1, [a#120,a#100], we'd skip
    1101             :                         // past all of the files in L2. When considering the second file in
    1102             :                         // L1, we'd improperly conclude that the second file overlaps
    1103             :                         // nothing in the second level and is cheap to compact, when in
    1104             :                         // reality we'd need to expand the compaction to include all 5
    1105             :                         // files.
    1106           1 :                         if sstableKeyCompare(cmp, outputFile.Largest, f.Largest) > 0 {
    1107           1 :                                 break
    1108             :                         }
    1109           1 :                         outputFile = outputIter.Next()
    1110             :                 }
    1111             : 
    1112             :                 // If the input level file or one of the overlapping files is
    1113             :                 // compacting, we're not going to be able to compact this file
    1114             :                 // anyways, so skip it.
    1115           1 :                 if compacting {
    1116           1 :                         continue
    1117             :                 }
    1118             : 
    1119           1 :                 compSz := compensatedSize(f) + responsibleForGarbageBytes(virtualBackings, f)
    1120           1 :                 scaledRatio := overlappingBytes * 1024 / compSz
    1121           1 :                 if scaledRatio < smallestRatio {
    1122           1 :                         smallestRatio = scaledRatio
    1123           1 :                         file = startIter.Take()
    1124           1 :                 }
    1125             :         }
    1126           1 :         return file, file.FileMetadata != nil
    1127             : }
    1128             : 
    1129             : // responsibleForGarbageBytes returns the amount of garbage in the backing
    1130             : // sstable that we consider the responsibility of this virtual sstable. For
    1131             : // non-virtual sstables, this is of course 0. For virtual sstables, we equally
    1132             : // distribute the responsibility of the garbage across all the virtual
    1133             : // sstables that are referencing the same backing sstable. One could
    1134             : // alternatively distribute this in proportion to the virtual sst sizes, but
    1135             : // it isn't clear that more sophisticated heuristics are worth it, given that
    1136             : // the garbage cannot be reclaimed until all the referencing virtual sstables
    1137             : // are compacted.
    1138           1 : func responsibleForGarbageBytes(virtualBackings *manifest.VirtualBackings, m *fileMetadata) uint64 {
    1139           1 :         if !m.Virtual {
    1140           1 :                 return 0
    1141           1 :         }
    1142           1 :         useCount, virtualizedSize := virtualBackings.Usage(m.FileBacking.DiskFileNum)
    1143           1 :         // Since virtualizedSize is the sum of the estimated size of all virtual
    1144           1 :         // ssts, we allow for the possibility that virtualizedSize could exceed
    1145           1 :         // m.FileBacking.Size.
    1146           1 :         totalGarbage := int64(m.FileBacking.Size) - int64(virtualizedSize)
    1147           1 :         if totalGarbage <= 0 {
    1148           1 :                 return 0
    1149           1 :         }
    1150           1 :         if useCount == 0 {
    1151           0 :                 // This cannot happen if m exists in the latest version. The call to
    1152           0 :                 // ResponsibleForGarbageBytes during compaction picking ensures that m
    1153           0 :                 // exists in the latest version by holding versionSet.logLock.
    1154           0 :                 panic(errors.AssertionFailedf("%s has zero useCount", m.String()))
    1155             :         }
    1156           1 :         return uint64(totalGarbage) / uint64(useCount)
    1157             : }
    1158             : 
    1159             : // pickAuto picks the best compaction, if any.
    1160             : //
    1161             : // On each call, pickAuto computes per-level size adjustments based on
    1162             : // in-progress compactions, and computes a per-level score. The levels are
    1163             : // iterated over in decreasing score order trying to find a valid compaction
    1164             : // anchored at that level.
    1165             : //
    1166             : // If a score-based compaction cannot be found, pickAuto falls back to looking
    1167             : // for an elision-only compaction to remove obsolete keys.
    1168           1 : func (p *compactionPickerByScore) pickAuto(env compactionEnv) (pc *pickedCompaction) {
    1169           1 :         // Compaction concurrency is controlled by L0 read-amp. We allow one
    1170           1 :         // additional compaction per L0CompactionConcurrency sublevels, as well as
    1171           1 :         // one additional compaction per CompactionDebtConcurrency bytes of
    1172           1 :         // compaction debt. Compaction concurrency is tied to L0 sublevels as that
    1173           1 :         // signal is independent of the database size. We tack on the compaction
    1174           1 :         // debt as a second signal to prevent compaction concurrency from dropping
    1175           1 :         // significantly right after a base compaction finishes, and before those
    1176           1 :         // bytes have been compacted further down the LSM.
    1177           1 :         if n := len(env.inProgressCompactions); n > 0 {
    1178           1 :                 l0ReadAmp := p.vers.L0Sublevels.MaxDepthAfterOngoingCompactions()
    1179           1 :                 compactionDebt := p.estimatedCompactionDebt(0)
    1180           1 :                 ccSignal1 := n * p.opts.Experimental.L0CompactionConcurrency
    1181           1 :                 ccSignal2 := uint64(n) * p.opts.Experimental.CompactionDebtConcurrency
    1182           1 :                 if l0ReadAmp < ccSignal1 && compactionDebt < ccSignal2 {
    1183           1 :                         return nil
    1184           1 :                 }
    1185             :         }
    1186             : 
    1187           1 :         scores := p.calculateLevelScores(env.inProgressCompactions)
    1188           1 : 
    1189           1 :         // TODO(bananabrick): Either remove, or change this into an event sent to the
    1190           1 :         // EventListener.
    1191           1 :         logCompaction := func(pc *pickedCompaction) {
    1192           0 :                 var buf bytes.Buffer
    1193           0 :                 for i := 0; i < numLevels; i++ {
    1194           0 :                         if i != 0 && i < p.baseLevel {
    1195           0 :                                 continue
    1196             :                         }
    1197             : 
    1198           0 :                         var info *candidateLevelInfo
    1199           0 :                         for j := range scores {
    1200           0 :                                 if scores[j].level == i {
    1201           0 :                                         info = &scores[j]
    1202           0 :                                         break
    1203             :                                 }
    1204             :                         }
    1205             : 
    1206           0 :                         marker := " "
    1207           0 :                         if pc.startLevel.level == info.level {
    1208           0 :                                 marker = "*"
    1209           0 :                         }
    1210           0 :                         fmt.Fprintf(&buf, "  %sL%d: %5.1f  %5.1f  %5.1f  %5.1f %8s  %8s",
    1211           0 :                                 marker, info.level, info.compensatedScoreRatio, info.compensatedScore,
    1212           0 :                                 info.uncompensatedScoreRatio, info.uncompensatedScore,
    1213           0 :                                 humanize.Bytes.Int64(int64(totalCompensatedSize(
    1214           0 :                                         p.vers.Levels[info.level].Iter(),
    1215           0 :                                 ))),
    1216           0 :                                 humanize.Bytes.Int64(p.levelMaxBytes[info.level]),
    1217           0 :                         )
    1218           0 : 
    1219           0 :                         count := 0
    1220           0 :                         for i := range env.inProgressCompactions {
    1221           0 :                                 c := &env.inProgressCompactions[i]
    1222           0 :                                 if c.inputs[0].level != info.level {
    1223           0 :                                         continue
    1224             :                                 }
    1225           0 :                                 count++
    1226           0 :                                 if count == 1 {
    1227           0 :                                         fmt.Fprintf(&buf, "  [")
    1228           0 :                                 } else {
    1229           0 :                                         fmt.Fprintf(&buf, " ")
    1230           0 :                                 }
    1231           0 :                                 fmt.Fprintf(&buf, "L%d->L%d", c.inputs[0].level, c.outputLevel)
    1232             :                         }
    1233           0 :                         if count > 0 {
    1234           0 :                                 fmt.Fprintf(&buf, "]")
    1235           0 :                         }
    1236           0 :                         fmt.Fprintf(&buf, "\n")
    1237             :                 }
    1238           0 :                 p.opts.Logger.Infof("pickAuto: L%d->L%d\n%s",
    1239           0 :                         pc.startLevel.level, pc.outputLevel.level, buf.String())
    1240             :         }
    1241             : 
    1242             :         // Check for a score-based compaction. candidateLevelInfos are first sorted
    1243             :         // by whether they should be compacted, so if we find a level which shouldn't
    1244             :         // be compacted, we can break early.
    1245           1 :         for i := range scores {
    1246           1 :                 info := &scores[i]
    1247           1 :                 if !info.shouldCompact() {
    1248           1 :                         break
    1249             :                 }
    1250           1 :                 if info.level == numLevels-1 {
    1251           1 :                         continue
    1252             :                 }
    1253             : 
    1254           1 :                 if info.level == 0 {
    1255           1 :                         pc = pickL0(env, p.opts, p.vers, p.baseLevel)
    1256           1 :                         // Fail-safe to protect against compacting the same sstable
    1257           1 :                         // concurrently.
    1258           1 :                         if pc != nil && !inputRangeAlreadyCompacting(env, pc) {
    1259           1 :                                 p.addScoresToPickedCompactionMetrics(pc, scores)
    1260           1 :                                 pc.score = info.compensatedScoreRatio
    1261           1 :                                 // TODO(bananabrick): Create an EventListener for logCompaction.
    1262           1 :                                 if false {
    1263           0 :                                         logCompaction(pc)
    1264           0 :                                 }
    1265           1 :                                 return pc
    1266             :                         }
    1267           1 :                         continue
    1268             :                 }
    1269             : 
    1270             :                 // info.level > 0
    1271           1 :                 var ok bool
    1272           1 :                 info.file, ok = pickCompactionSeedFile(p.vers, p.virtualBackings, p.opts, info.level, info.outputLevel, env.earliestSnapshotSeqNum)
    1273           1 :                 if !ok {
    1274           1 :                         continue
    1275             :                 }
    1276             : 
    1277           1 :                 pc := pickAutoLPositive(env, p.opts, p.vers, *info, p.baseLevel)
    1278           1 :                 // Fail-safe to protect against compacting the same sstable concurrently.
    1279           1 :                 if pc != nil && !inputRangeAlreadyCompacting(env, pc) {
    1280           1 :                         p.addScoresToPickedCompactionMetrics(pc, scores)
    1281           1 :                         pc.score = info.compensatedScoreRatio
    1282           1 :                         // TODO(bananabrick): Create an EventListener for logCompaction.
    1283           1 :                         if false {
    1284           0 :                                 logCompaction(pc)
    1285           0 :                         }
    1286           1 :                         return pc
    1287             :                 }
    1288             :         }
    1289             : 
    1290             :         // Check for files which contain excessive point tombstones that could slow
    1291             :         // down reads. Unlike elision-only compactions, these compactions may select
    1292             :         // a file at any level rather than only the lowest level.
    1293           1 :         if pc := p.pickTombstoneDensityCompaction(env); pc != nil {
    1294           1 :                 return pc
    1295           1 :         }
    1296             : 
    1297             :         // Check for L6 files with tombstones that may be elided. These files may
    1298             :         // exist if a snapshot prevented the elision of a tombstone or because of
    1299             :         // a move compaction. These are low-priority compactions because they
    1300             :         // don't help us keep up with writes, just reclaim disk space.
    1301           1 :         if pc := p.pickElisionOnlyCompaction(env); pc != nil {
    1302           1 :                 return pc
    1303           1 :         }
    1304             : 
    1305           1 :         if pc := p.pickReadTriggeredCompaction(env); pc != nil {
    1306           0 :                 return pc
    1307           0 :         }
    1308             : 
    1309             :         // NB: This should only be run if a read compaction wasn't
    1310             :         // scheduled.
    1311             :         //
    1312             :         // We won't be scheduling a read compaction right now, and in
    1313             :         // read heavy workloads, compactions won't be scheduled frequently
    1314             :         // because flushes aren't frequent. So we need to signal to the
    1315             :         // iterator to schedule a compaction when it adds compactions to
    1316             :         // the read compaction queue.
    1317             :         //
    1318             :         // We need the nil check here because without it, we have some
    1319             :         // tests which don't set that variable fail. Since there's a
    1320             :         // chance that one of those tests wouldn't want extra compactions
    1321             :         // to be scheduled, I added this check here, instead of
    1322             :         // setting rescheduleReadCompaction in those tests.
    1323           1 :         if env.readCompactionEnv.rescheduleReadCompaction != nil {
    1324           1 :                 *env.readCompactionEnv.rescheduleReadCompaction = true
    1325           1 :         }
    1326             : 
    1327             :         // At the lowest possible compaction-picking priority, look for files marked
    1328             :         // for compaction. Pebble will mark files for compaction if they have atomic
    1329             :         // compaction units that span multiple files. While current Pebble code does
    1330             :         // not construct such sstables, RocksDB and earlier versions of Pebble may
    1331             :         // have created them. These split user keys form sets of files that must be
    1332             :         // compacted together for correctness (referred to as "atomic compaction
    1333             :         // units" within the code). Rewrite them in-place.
    1334             :         //
    1335             :         // It's also possible that a file may have been marked for compaction by
    1336             :         // even earlier versions of Pebble code, since FileMetadata's
    1337             :         // MarkedForCompaction field is persisted in the manifest. That's okay. We
    1338             :         // previously would've ignored the designation, whereas now we'll re-compact
    1339             :         // the file in place.
    1340           1 :         if p.vers.Stats.MarkedForCompaction > 0 {
    1341           0 :                 if pc := p.pickRewriteCompaction(env); pc != nil {
    1342           0 :                         return pc
    1343           0 :                 }
    1344             :         }
    1345             : 
    1346           1 :         return nil
    1347             : }
    1348             : 
    1349             : func (p *compactionPickerByScore) addScoresToPickedCompactionMetrics(
    1350             :         pc *pickedCompaction, candInfo [numLevels]candidateLevelInfo,
    1351           1 : ) {
    1352           1 : 
    1353           1 :         // candInfo is sorted by score, not by compaction level.
    1354           1 :         infoByLevel := [numLevels]candidateLevelInfo{}
    1355           1 :         for i := range candInfo {
    1356           1 :                 level := candInfo[i].level
    1357           1 :                 infoByLevel[level] = candInfo[i]
    1358           1 :         }
    1359             :         // Gather the compaction scores for the levels participating in the compaction.
    1360           1 :         pc.pickerMetrics.scores = make([]float64, len(pc.inputs))
    1361           1 :         inputIdx := 0
    1362           1 :         for i := range infoByLevel {
    1363           1 :                 if pc.inputs[inputIdx].level == infoByLevel[i].level {
    1364           1 :                         pc.pickerMetrics.scores[inputIdx] = infoByLevel[i].compensatedScoreRatio
    1365           1 :                         inputIdx++
    1366           1 :                 }
    1367           1 :                 if inputIdx == len(pc.inputs) {
    1368           1 :                         break
    1369             :                 }
    1370             :         }
    1371             : }
    1372             : 
    1373             : // elisionOnlyAnnotator is a manifest.Annotator that annotates B-Tree
    1374             : // nodes with the *fileMetadata of a file meeting the obsolete keys criteria
    1375             : // for an elision-only compaction within the subtree. If multiple files meet
    1376             : // the criteria, it chooses whichever file has the lowest LargestSeqNum. The
    1377             : // lowest LargestSeqNum file will be the first eligible for an elision-only
    1378             : // compaction once snapshots less than or equal to its LargestSeqNum are closed.
    1379             : var elisionOnlyAnnotator = &manifest.Annotator[fileMetadata]{
    1380             :         Aggregator: manifest.PickFileAggregator{
    1381           1 :                 Filter: func(f *fileMetadata) (eligible bool, cacheOK bool) {
    1382           1 :                         if f.IsCompacting() {
    1383           1 :                                 return false, true
    1384           1 :                         }
    1385           1 :                         if !f.StatsValid() {
    1386           1 :                                 return false, false
    1387           1 :                         }
    1388             :                         // Bottommost files are large and not worthwhile to compact just
    1389             :                         // to remove a few tombstones. Consider a file eligible only if
    1390             :                         // either its own range deletions delete at least 10% of its data or
    1391             :                         // its deletion tombstones make at least 10% of its entries.
    1392             :                         //
    1393             :                         // TODO(jackson): This does not account for duplicate user keys
    1394             :                         // which may be collapsed. Ideally, we would have 'obsolete keys'
    1395             :                         // statistics that would include tombstones, the keys that are
    1396             :                         // dropped by tombstones and duplicated user keys. See #847.
    1397             :                         //
    1398             :                         // Note that tables that contain exclusively range keys (i.e. no point keys,
    1399             :                         // `NumEntries` and `RangeDeletionsBytesEstimate` are both zero) are excluded
    1400             :                         // from elision-only compactions.
    1401             :                         // TODO(travers): Consider an alternative heuristic for elision of range-keys.
    1402           1 :                         return f.Stats.RangeDeletionsBytesEstimate*10 >= f.Size || f.Stats.NumDeletions*10 > f.Stats.NumEntries, true
    1403             :                 },
    1404           1 :                 Compare: func(f1 *fileMetadata, f2 *fileMetadata) bool {
    1405           1 :                         return f1.LargestSeqNum < f2.LargestSeqNum
    1406           1 :                 },
    1407             :         },
    1408             : }
    1409             : 
    1410             : // markedForCompactionAnnotator is a manifest.Annotator that annotates B-Tree
    1411             : // nodes with the *fileMetadata of a file that is marked for compaction
    1412             : // within the subtree. If multiple files meet the criteria, it chooses
    1413             : // whichever file has the lowest LargestSeqNum.
    1414             : var markedForCompactionAnnotator = &manifest.Annotator[fileMetadata]{
    1415             :         Aggregator: manifest.PickFileAggregator{
    1416           0 :                 Filter: func(f *fileMetadata) (eligible bool, cacheOK bool) {
    1417           0 :                         return f.MarkedForCompaction, true
    1418           0 :                 },
    1419           0 :                 Compare: func(f1 *fileMetadata, f2 *fileMetadata) bool {
    1420           0 :                         return f1.LargestSeqNum < f2.LargestSeqNum
    1421           0 :                 },
    1422             :         },
    1423             : }
    1424             : 
    1425             : // pickedCompactionFromCandidateFile creates a pickedCompaction from a *fileMetadata
    1426             : // with various checks to ensure that the file still exists in the expected level
    1427             : // and isn't already being compacted.
    1428             : func (p *compactionPickerByScore) pickedCompactionFromCandidateFile(
    1429             :         candidate *fileMetadata, env compactionEnv, startLevel int, outputLevel int, kind compactionKind,
    1430           1 : ) *pickedCompaction {
    1431           1 :         if candidate == nil || candidate.IsCompacting() {
    1432           1 :                 return nil
    1433           1 :         }
    1434             : 
    1435           1 :         var inputs manifest.LevelSlice
    1436           1 :         if startLevel == 0 {
    1437           1 :                 // Overlapping L0 files must also be compacted alongside the candidate.
    1438           1 :                 inputs = p.vers.Overlaps(0, candidate.UserKeyBounds())
    1439           1 :         } else {
    1440           1 :                 inputs = p.vers.Levels[startLevel].Find(p.opts.Comparer.Compare, candidate)
    1441           1 :         }
    1442           1 :         if invariants.Enabled {
    1443           1 :                 found := false
    1444           1 :                 inputs.Each(func(f *fileMetadata) {
    1445           1 :                         if f.FileNum == candidate.FileNum {
    1446           1 :                                 found = true
    1447           1 :                         }
    1448             :                 })
    1449           1 :                 if !found {
    1450           0 :                         panic(fmt.Sprintf("file %s not found in level %d as expected", candidate.FileNum, startLevel))
    1451             :                 }
    1452             :         }
    1453             : 
    1454           1 :         pc := newPickedCompaction(p.opts, p.vers, startLevel, outputLevel, p.baseLevel)
    1455           1 :         pc.kind = kind
    1456           1 :         pc.startLevel.files = inputs
    1457           1 :         pc.smallest, pc.largest = manifest.KeyRange(pc.cmp, pc.startLevel.files.Iter())
    1458           1 : 
    1459           1 :         // Fail-safe to protect against compacting the same sstable concurrently.
    1460           1 :         if inputRangeAlreadyCompacting(env, pc) {
    1461           1 :                 return nil
    1462           1 :         }
    1463             : 
    1464           1 :         if !pc.setupInputs(p.opts, env.diskAvailBytes, pc.startLevel) {
    1465           1 :                 return nil
    1466           1 :         }
    1467             : 
    1468           1 :         return pc
    1469             : }
    1470             : 
    1471             : // pickElisionOnlyCompaction looks for compactions of sstables in the
    1472             : // bottommost level containing obsolete records that may now be dropped.
    1473             : func (p *compactionPickerByScore) pickElisionOnlyCompaction(
    1474             :         env compactionEnv,
    1475           1 : ) (pc *pickedCompaction) {
    1476           1 :         if p.opts.private.disableElisionOnlyCompactions {
    1477           1 :                 return nil
    1478           1 :         }
    1479           1 :         candidate := elisionOnlyAnnotator.LevelAnnotation(p.vers.Levels[numLevels-1])
    1480           1 :         if candidate == nil {
    1481           1 :                 return nil
    1482           1 :         }
    1483           1 :         if candidate.LargestSeqNum >= env.earliestSnapshotSeqNum {
    1484           1 :                 return nil
    1485           1 :         }
    1486           1 :         return p.pickedCompactionFromCandidateFile(candidate, env, numLevels-1, numLevels-1, compactionKindElisionOnly)
    1487             : }
    1488             : 
    1489             : // pickRewriteCompaction attempts to construct a compaction that
    1490             : // rewrites a file marked for compaction. pickRewriteCompaction will
    1491             : // pull in adjacent files in the file's atomic compaction unit if
    1492             : // necessary. A rewrite compaction outputs files to the same level as
    1493             : // the input level.
    1494           0 : func (p *compactionPickerByScore) pickRewriteCompaction(env compactionEnv) (pc *pickedCompaction) {
    1495           0 :         for l := numLevels - 1; l >= 0; l-- {
    1496           0 :                 candidate := markedForCompactionAnnotator.LevelAnnotation(p.vers.Levels[l])
    1497           0 :                 if candidate == nil {
    1498           0 :                         // Try the next level.
    1499           0 :                         continue
    1500             :                 }
    1501           0 :                 pc := p.pickedCompactionFromCandidateFile(candidate, env, l, l, compactionKindRewrite)
    1502           0 :                 if pc != nil {
    1503           0 :                         return pc
    1504           0 :                 }
    1505             :         }
    1506           0 :         return nil
    1507             : }
    1508             : 
    1509           1 : func (p *compactionPickerByScore) initTombstoneDensityAnnotator(opts *Options) {
    1510           1 :         p.tombstoneDensityAnnotator = &manifest.Annotator[fileMetadata]{
    1511           1 :                 Aggregator: manifest.PickFileAggregator{
    1512           1 :                         Filter: func(f *fileMetadata) (eligible bool, cacheOK bool) {
    1513           1 :                                 if f.IsCompacting() {
    1514           1 :                                         return false, true
    1515           1 :                                 }
    1516           1 :                                 if !f.StatsValid() {
    1517           1 :                                         return false, false
    1518           1 :                                 }
    1519           1 :                                 return f.Stats.TombstoneDenseBlocksRatio > opts.Experimental.TombstoneDenseCompactionThreshold, true
    1520             :                         },
    1521           1 :                         Compare: func(a, b *fileMetadata) bool {
    1522           1 :                                 return a.Stats.TombstoneDenseBlocksRatio > b.Stats.TombstoneDenseBlocksRatio
    1523           1 :                         },
    1524             :                 },
    1525             :         }
    1526             : }
    1527             : 
    1528             : // pickTombstoneDensityCompaction looks for a compaction that eliminates
    1529             : // regions of extremely high point tombstone density. For each level, it picks
    1530             : // a file where the ratio of tombstone-dense blocks is at least
    1531             : // options.Experimental.MinTombstoneDenseRatio, prioritizing compaction of
    1532             : // files with higher ratios of tombstone-dense blocks.
    1533             : func (p *compactionPickerByScore) pickTombstoneDensityCompaction(
    1534             :         env compactionEnv,
    1535           1 : ) (pc *pickedCompaction) {
    1536           1 :         if p.opts.Experimental.TombstoneDenseCompactionThreshold <= 0 {
    1537           0 :                 // Tombstone density compactions are disabled.
    1538           0 :                 return nil
    1539           0 :         }
    1540             : 
    1541           1 :         var candidate *fileMetadata
    1542           1 :         var level int
    1543           1 :         // NB: we don't consider the lowest level because elision-only compactions
    1544           1 :         // handle that case.
    1545           1 :         for l := 0; l < numLevels-1; l++ {
    1546           1 :                 f := p.tombstoneDensityAnnotator.LevelAnnotation(p.vers.Levels[l])
    1547           1 :                 newCandidate := p.tombstoneDensityAnnotator.Aggregator.Merge(f, candidate)
    1548           1 :                 if newCandidate != candidate {
    1549           1 :                         candidate = newCandidate
    1550           1 :                         level = l
    1551           1 :                 }
    1552             :         }
    1553             : 
    1554           1 :         return p.pickedCompactionFromCandidateFile(candidate, env, level, defaultOutputLevel(level, p.baseLevel), compactionKindTombstoneDensity)
    1555             : }
    1556             : 
    1557             : // pickAutoLPositive picks an automatic compaction for the candidate
    1558             : // file in a positive-numbered level. This function must not be used for
    1559             : // L0.
    1560             : func pickAutoLPositive(
    1561             :         env compactionEnv, opts *Options, vers *version, cInfo candidateLevelInfo, baseLevel int,
    1562           1 : ) (pc *pickedCompaction) {
    1563           1 :         if cInfo.level == 0 {
    1564           0 :                 panic("pebble: pickAutoLPositive called for L0")
    1565             :         }
    1566             : 
    1567           1 :         pc = newPickedCompaction(opts, vers, cInfo.level, defaultOutputLevel(cInfo.level, baseLevel), baseLevel)
    1568           1 :         if pc.outputLevel.level != cInfo.outputLevel {
    1569           0 :                 panic("pebble: compaction picked unexpected output level")
    1570             :         }
    1571           1 :         pc.startLevel.files = cInfo.file.Slice()
    1572           1 :         // Files in level 0 may overlap each other, so pick up all overlapping ones.
    1573           1 :         if pc.startLevel.level == 0 {
    1574           0 :                 cmp := opts.Comparer.Compare
    1575           0 :                 smallest, largest := manifest.KeyRange(cmp, pc.startLevel.files.Iter())
    1576           0 :                 pc.startLevel.files = vers.Overlaps(0, base.UserKeyBoundsFromInternal(smallest, largest))
    1577           0 :                 if pc.startLevel.files.Empty() {
    1578           0 :                         panic("pebble: empty compaction")
    1579             :                 }
    1580             :         }
    1581             : 
    1582           1 :         if !pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
    1583           0 :                 return nil
    1584           0 :         }
    1585           1 :         return pc.maybeAddLevel(opts, env.diskAvailBytes)
    1586             : }
    1587             : 
    1588             : // maybeAddLevel maybe adds a level to the picked compaction.
    1589           1 : func (pc *pickedCompaction) maybeAddLevel(opts *Options, diskAvailBytes uint64) *pickedCompaction {
    1590           1 :         pc.pickerMetrics.singleLevelOverlappingRatio = pc.overlappingRatio()
    1591           1 :         if pc.outputLevel.level == numLevels-1 {
    1592           1 :                 // Don't add a level if the current output level is in L6
    1593           1 :                 return pc
    1594           1 :         }
    1595           1 :         if !opts.Experimental.MultiLevelCompactionHeuristic.allowL0() && pc.startLevel.level == 0 {
    1596           1 :                 return pc
    1597           1 :         }
    1598           1 :         if pc.compactionSize() > expandedCompactionByteSizeLimit(
    1599           1 :                 opts, adjustedOutputLevel(pc.outputLevel.level, pc.baseLevel), diskAvailBytes) {
    1600           1 :                 // Don't add a level if the current compaction exceeds the compaction size limit
    1601           1 :                 return pc
    1602           1 :         }
    1603           1 :         return opts.Experimental.MultiLevelCompactionHeuristic.pick(pc, opts, diskAvailBytes)
    1604             : }
    1605             : 
    1606             : // MultiLevelHeuristic evaluates whether to add files from the next level into the compaction.
    1607             : type MultiLevelHeuristic interface {
    1608             :         // Evaluate returns the preferred compaction.
    1609             :         pick(pc *pickedCompaction, opts *Options, diskAvailBytes uint64) *pickedCompaction
    1610             : 
    1611             :         // Returns if the heuristic allows L0 to be involved in ML compaction
    1612             :         allowL0() bool
    1613             : 
    1614             :         // String implements fmt.Stringer.
    1615             :         String() string
    1616             : }
    1617             : 
    1618             : // NoMultiLevel will never add an additional level to the compaction.
    1619             : type NoMultiLevel struct{}
    1620             : 
    1621             : var _ MultiLevelHeuristic = (*NoMultiLevel)(nil)
    1622             : 
    1623             : func (nml NoMultiLevel) pick(
    1624             :         pc *pickedCompaction, opts *Options, diskAvailBytes uint64,
    1625           1 : ) *pickedCompaction {
    1626           1 :         return pc
    1627           1 : }
    1628             : 
    1629           1 : func (nml NoMultiLevel) allowL0() bool  { return false }
    1630           1 : func (nml NoMultiLevel) String() string { return "none" }
    1631             : 
    1632           1 : func (pc *pickedCompaction) predictedWriteAmp() float64 {
    1633           1 :         var bytesToCompact uint64
    1634           1 :         var higherLevelBytes uint64
    1635           1 :         for i := range pc.inputs {
    1636           1 :                 levelSize := pc.inputs[i].files.SizeSum()
    1637           1 :                 bytesToCompact += levelSize
    1638           1 :                 if i != len(pc.inputs)-1 {
    1639           1 :                         higherLevelBytes += levelSize
    1640           1 :                 }
    1641             :         }
    1642           1 :         return float64(bytesToCompact) / float64(higherLevelBytes)
    1643             : }
    1644             : 
    1645           1 : func (pc *pickedCompaction) overlappingRatio() float64 {
    1646           1 :         var higherLevelBytes uint64
    1647           1 :         var lowestLevelBytes uint64
    1648           1 :         for i := range pc.inputs {
    1649           1 :                 levelSize := pc.inputs[i].files.SizeSum()
    1650           1 :                 if i == len(pc.inputs)-1 {
    1651           1 :                         lowestLevelBytes += levelSize
    1652           1 :                         continue
    1653             :                 }
    1654           1 :                 higherLevelBytes += levelSize
    1655             :         }
    1656           1 :         return float64(lowestLevelBytes) / float64(higherLevelBytes)
    1657             : }
    1658             : 
    1659             : // WriteAmpHeuristic defines a multi level compaction heuristic which will add
    1660             : // an additional level to the picked compaction if it reduces predicted write
    1661             : // amp of the compaction + the addPropensity constant.
    1662             : type WriteAmpHeuristic struct {
    1663             :         // addPropensity is a constant that affects the propensity to conduct multilevel
    1664             :         // compactions. If positive, a multilevel compaction may get picked even if
    1665             :         // the single level compaction has lower write amp, and vice versa.
    1666             :         AddPropensity float64
    1667             : 
    1668             :         // AllowL0 if true, allow l0 to be involved in a ML compaction.
    1669             :         AllowL0 bool
    1670             : }
    1671             : 
    1672             : var _ MultiLevelHeuristic = (*WriteAmpHeuristic)(nil)
    1673             : 
    1674             : // TODO(msbutler): microbenchmark the extent to which multilevel compaction
    1675             : // picking slows down the compaction picking process.  This should be as fast as
    1676             : // possible since Compaction-picking holds d.mu, which prevents WAL rotations,
    1677             : // in-progress flushes and compactions from completing, etc. Consider ways to
    1678             : // deduplicate work, given that setupInputs has already been called.
    1679             : func (wa WriteAmpHeuristic) pick(
    1680             :         pcOrig *pickedCompaction, opts *Options, diskAvailBytes uint64,
    1681           1 : ) *pickedCompaction {
    1682           1 :         pcMulti := pcOrig.clone()
    1683           1 :         if !pcMulti.setupMultiLevelCandidate(opts, diskAvailBytes) {
    1684           1 :                 return pcOrig
    1685           1 :         }
    1686             :         // We consider the addition of a level as an "expansion" of the compaction.
    1687             :         // If pcMulti is past the expanded compaction byte size limit already,
    1688             :         // we don't consider it.
    1689           1 :         if pcMulti.compactionSize() >= expandedCompactionByteSizeLimit(
    1690           1 :                 opts, adjustedOutputLevel(pcMulti.outputLevel.level, pcMulti.baseLevel), diskAvailBytes) {
    1691           1 :                 return pcOrig
    1692           1 :         }
    1693           1 :         picked := pcOrig
    1694           1 :         if pcMulti.predictedWriteAmp() <= pcOrig.predictedWriteAmp()+wa.AddPropensity {
    1695           1 :                 picked = pcMulti
    1696           1 :         }
    1697             :         // Regardless of what compaction was picked, log the multilevelOverlapping ratio.
    1698           1 :         picked.pickerMetrics.multiLevelOverlappingRatio = pcMulti.overlappingRatio()
    1699           1 :         return picked
    1700             : }
    1701             : 
    1702           1 : func (wa WriteAmpHeuristic) allowL0() bool {
    1703           1 :         return wa.AllowL0
    1704           1 : }
    1705             : 
    1706             : // String implements fmt.Stringer.
    1707           1 : func (wa WriteAmpHeuristic) String() string {
    1708           1 :         return fmt.Sprintf("wamp(%.2f, %t)", wa.AddPropensity, wa.AllowL0)
    1709           1 : }
    1710             : 
    1711             : // Helper method to pick compactions originating from L0. Uses information about
    1712             : // sublevels to generate a compaction.
    1713           1 : func pickL0(env compactionEnv, opts *Options, vers *version, baseLevel int) (pc *pickedCompaction) {
    1714           1 :         // It is important to pass information about Lbase files to L0Sublevels
    1715           1 :         // so it can pick a compaction that does not conflict with an Lbase => Lbase+1
    1716           1 :         // compaction. Without this, we observed reduced concurrency of L0=>Lbase
    1717           1 :         // compactions, and increasing read amplification in L0.
    1718           1 :         //
    1719           1 :         // TODO(bilal) Remove the minCompactionDepth parameter once fixing it at 1
    1720           1 :         // has been shown to not cause a performance regression.
    1721           1 :         lcf, err := vers.L0Sublevels.PickBaseCompaction(1, vers.Levels[baseLevel].Slice())
    1722           1 :         if err != nil {
    1723           0 :                 opts.Logger.Errorf("error when picking base compaction: %s", err)
    1724           0 :                 return
    1725           0 :         }
    1726           1 :         if lcf != nil {
    1727           1 :                 pc = newPickedCompactionFromL0(lcf, opts, vers, baseLevel, true)
    1728           1 :                 pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel)
    1729           1 :                 if pc.startLevel.files.Empty() {
    1730           0 :                         opts.Logger.Fatalf("empty compaction chosen")
    1731           0 :                 }
    1732           1 :                 return pc.maybeAddLevel(opts, env.diskAvailBytes)
    1733             :         }
    1734             : 
    1735             :         // Couldn't choose a base compaction. Try choosing an intra-L0
    1736             :         // compaction. Note that we pass in L0CompactionThreshold here as opposed to
    1737             :         // 1, since choosing a single sublevel intra-L0 compaction is
    1738             :         // counterproductive.
    1739           1 :         lcf, err = vers.L0Sublevels.PickIntraL0Compaction(env.earliestUnflushedSeqNum, minIntraL0Count)
    1740           1 :         if err != nil {
    1741           0 :                 opts.Logger.Errorf("error when picking intra-L0 compaction: %s", err)
    1742           0 :                 return
    1743           0 :         }
    1744           1 :         if lcf != nil {
    1745           1 :                 pc = newPickedCompactionFromL0(lcf, opts, vers, 0, false)
    1746           1 :                 if !pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
    1747           0 :                         return nil
    1748           0 :                 }
    1749           1 :                 if pc.startLevel.files.Empty() {
    1750           0 :                         opts.Logger.Fatalf("empty compaction chosen")
    1751           0 :                 }
    1752           1 :                 {
    1753           1 :                         iter := pc.startLevel.files.Iter()
    1754           1 :                         if iter.First() == nil || iter.Next() == nil {
    1755           0 :                                 // A single-file intra-L0 compaction is unproductive.
    1756           0 :                                 return nil
    1757           0 :                         }
    1758             :                 }
    1759             : 
    1760           1 :                 pc.smallest, pc.largest = manifest.KeyRange(pc.cmp, pc.startLevel.files.Iter())
    1761             :         }
    1762           1 :         return pc
    1763             : }
    1764             : 
    1765             : func pickManualCompaction(
    1766             :         vers *version, opts *Options, env compactionEnv, baseLevel int, manual *manualCompaction,
    1767           1 : ) (pc *pickedCompaction, retryLater bool) {
    1768           1 :         outputLevel := manual.level + 1
    1769           1 :         if manual.level == 0 {
    1770           1 :                 outputLevel = baseLevel
    1771           1 :         } else if manual.level < baseLevel {
    1772           1 :                 // The start level for a compaction must be >= Lbase. A manual
    1773           1 :                 // compaction could have been created adhering to that condition, and
    1774           1 :                 // then an automatic compaction came in and compacted all of the
    1775           1 :                 // sstables in Lbase to Lbase+1 which caused Lbase to change. Simply
    1776           1 :                 // ignore this manual compaction as there is nothing to do (manual.level
    1777           1 :                 // points to an empty level).
    1778           1 :                 return nil, false
    1779           1 :         }
    1780             :         // This conflictsWithInProgress call is necessary for the manual compaction to
    1781             :         // be retried when it conflicts with an ongoing automatic compaction. Without
    1782             :         // it, the compaction is dropped due to pc.setupInputs returning false since
    1783             :         // the input/output range is already being compacted, and the manual
    1784             :         // compaction ends with a non-compacted LSM.
    1785           1 :         if conflictsWithInProgress(manual, outputLevel, env.inProgressCompactions, opts.Comparer.Compare) {
    1786           1 :                 return nil, true
    1787           1 :         }
    1788           1 :         pc = newPickedCompaction(opts, vers, manual.level, defaultOutputLevel(manual.level, baseLevel), baseLevel)
    1789           1 :         manual.outputLevel = pc.outputLevel.level
    1790           1 :         pc.startLevel.files = vers.Overlaps(manual.level, base.UserKeyBoundsInclusive(manual.start, manual.end))
    1791           1 :         if pc.startLevel.files.Empty() {
    1792           1 :                 // Nothing to do
    1793           1 :                 return nil, false
    1794           1 :         }
    1795           1 :         if !pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
    1796           1 :                 // setupInputs returned false indicating there's a conflicting
    1797           1 :                 // concurrent compaction.
    1798           1 :                 return nil, true
    1799           1 :         }
    1800           1 :         if pc = pc.maybeAddLevel(opts, env.diskAvailBytes); pc == nil {
    1801           0 :                 return nil, false
    1802           0 :         }
    1803           1 :         if pc.outputLevel.level != outputLevel {
    1804           1 :                 if len(pc.extraLevels) > 0 {
    1805           1 :                         // multilevel compactions relax this invariant
    1806           1 :                 } else {
    1807           0 :                         panic("pebble: compaction picked unexpected output level")
    1808             :                 }
    1809             :         }
    1810             :         // Fail-safe to protect against compacting the same sstable concurrently.
    1811           1 :         if inputRangeAlreadyCompacting(env, pc) {
    1812           1 :                 return nil, true
    1813           1 :         }
    1814           1 :         return pc, false
    1815             : }
    1816             : 
    1817             : // pickDownloadCompaction picks a download compaction for the downloadSpan,
    1818             : // which could be specified as being performed either by a copy compaction of
    1819             : // the backing file or a rewrite compaction.
    1820             : func pickDownloadCompaction(
    1821             :         vers *version,
    1822             :         opts *Options,
    1823             :         env compactionEnv,
    1824             :         baseLevel int,
    1825             :         kind compactionKind,
    1826             :         level int,
    1827             :         file *fileMetadata,
    1828           1 : ) (pc *pickedCompaction) {
    1829           1 :         // Check if the file is compacting already.
    1830           1 :         if file.CompactionState == manifest.CompactionStateCompacting {
    1831           0 :                 return nil
    1832           0 :         }
    1833           1 :         if kind != compactionKindCopy && kind != compactionKindRewrite {
    1834           0 :                 panic("invalid download/rewrite compaction kind")
    1835             :         }
    1836           1 :         pc = newPickedCompaction(opts, vers, level, level, baseLevel)
    1837           1 :         pc.kind = kind
    1838           1 :         pc.startLevel.files = manifest.NewLevelSliceKeySorted(opts.Comparer.Compare, []*fileMetadata{file})
    1839           1 :         if !pc.setupInputs(opts, env.diskAvailBytes, pc.startLevel) {
    1840           0 :                 // setupInputs returned false indicating there's a conflicting
    1841           0 :                 // concurrent compaction.
    1842           0 :                 return nil
    1843           0 :         }
    1844           1 :         if pc.outputLevel.level != level {
    1845           0 :                 panic("pebble: download compaction picked unexpected output level")
    1846             :         }
    1847             :         // Fail-safe to protect against compacting the same sstable concurrently.
    1848           1 :         if inputRangeAlreadyCompacting(env, pc) {
    1849           0 :                 return nil
    1850           0 :         }
    1851           1 :         return pc
    1852             : }
    1853             : 
    1854             : func (p *compactionPickerByScore) pickReadTriggeredCompaction(
    1855             :         env compactionEnv,
    1856           1 : ) (pc *pickedCompaction) {
    1857           1 :         // If a flush is in-progress or expected to happen soon, it means more writes are taking place. We would
    1858           1 :         // soon be scheduling more write focussed compactions. In this case, skip read compactions as they are
    1859           1 :         // lower priority.
    1860           1 :         if env.readCompactionEnv.flushing || env.readCompactionEnv.readCompactions == nil {
    1861           1 :                 return nil
    1862           1 :         }
    1863           1 :         for env.readCompactionEnv.readCompactions.size > 0 {
    1864           0 :                 rc := env.readCompactionEnv.readCompactions.remove()
    1865           0 :                 if pc = pickReadTriggeredCompactionHelper(p, rc, env); pc != nil {
    1866           0 :                         break
    1867             :                 }
    1868             :         }
    1869           1 :         return pc
    1870             : }
    1871             : 
    1872             : func pickReadTriggeredCompactionHelper(
    1873             :         p *compactionPickerByScore, rc *readCompaction, env compactionEnv,
    1874           0 : ) (pc *pickedCompaction) {
    1875           0 :         overlapSlice := p.vers.Overlaps(rc.level, base.UserKeyBoundsInclusive(rc.start, rc.end))
    1876           0 :         if overlapSlice.Empty() {
    1877           0 :                 // If there is no overlap, then the file with the key range
    1878           0 :                 // must have been compacted away. So, we don't proceed to
    1879           0 :                 // compact the same key range again.
    1880           0 :                 return nil
    1881           0 :         }
    1882             : 
    1883           0 :         iter := overlapSlice.Iter()
    1884           0 :         var fileMatches bool
    1885           0 :         for f := iter.First(); f != nil; f = iter.Next() {
    1886           0 :                 if f.FileNum == rc.fileNum {
    1887           0 :                         fileMatches = true
    1888           0 :                         break
    1889             :                 }
    1890             :         }
    1891           0 :         if !fileMatches {
    1892           0 :                 return nil
    1893           0 :         }
    1894             : 
    1895           0 :         pc = newPickedCompaction(p.opts, p.vers, rc.level, defaultOutputLevel(rc.level, p.baseLevel), p.baseLevel)
    1896           0 : 
    1897           0 :         pc.startLevel.files = overlapSlice
    1898           0 :         if !pc.setupInputs(p.opts, env.diskAvailBytes, pc.startLevel) {
    1899           0 :                 return nil
    1900           0 :         }
    1901           0 :         if inputRangeAlreadyCompacting(env, pc) {
    1902           0 :                 return nil
    1903           0 :         }
    1904           0 :         pc.kind = compactionKindRead
    1905           0 : 
    1906           0 :         // Prevent read compactions which are too wide.
    1907           0 :         outputOverlaps := pc.version.Overlaps(pc.outputLevel.level, pc.userKeyBounds())
    1908           0 :         if outputOverlaps.SizeSum() > pc.maxReadCompactionBytes {
    1909           0 :                 return nil
    1910           0 :         }
    1911             : 
    1912             :         // Prevent compactions which start with a small seed file X, but overlap
    1913             :         // with over allowedCompactionWidth * X file sizes in the output layer.
    1914           0 :         const allowedCompactionWidth = 35
    1915           0 :         if outputOverlaps.SizeSum() > overlapSlice.SizeSum()*allowedCompactionWidth {
    1916           0 :                 return nil
    1917           0 :         }
    1918             : 
    1919           0 :         return pc
    1920             : }
    1921             : 
    1922           0 : func (p *compactionPickerByScore) forceBaseLevel1() {
    1923           0 :         p.baseLevel = 1
    1924           0 : }
    1925             : 
    1926           1 : func inputRangeAlreadyCompacting(env compactionEnv, pc *pickedCompaction) bool {
    1927           1 :         for _, cl := range pc.inputs {
    1928           1 :                 iter := cl.files.Iter()
    1929           1 :                 for f := iter.First(); f != nil; f = iter.Next() {
    1930           1 :                         if f.IsCompacting() {
    1931           1 :                                 return true
    1932           1 :                         }
    1933             :                 }
    1934             :         }
    1935             : 
    1936             :         // Look for active compactions outputting to the same region of the key
    1937             :         // space in the same output level. Two potential compactions may conflict
    1938             :         // without sharing input files if there are no files in the output level
    1939             :         // that overlap with the intersection of the compactions' key spaces.
    1940             :         //
    1941             :         // Consider an active L0->Lbase compaction compacting two L0 files one
    1942             :         // [a-f] and the other [t-z] into Lbase.
    1943             :         //
    1944             :         // L0
    1945             :         //     ↦ 000100  ↤                           ↦  000101   ↤
    1946             :         // L1
    1947             :         //     ↦ 000004  ↤
    1948             :         //     a b c d e f g h i j k l m n o p q r s t u v w x y z
    1949             :         //
    1950             :         // If a new file 000102 [j-p] is flushed while the existing compaction is
    1951             :         // still ongoing, new file would not be in any compacting sublevel
    1952             :         // intervals and would not overlap with any Lbase files that are also
    1953             :         // compacting. However, this compaction cannot be picked because the
    1954             :         // compaction's output key space [j-p] would overlap the existing
    1955             :         // compaction's output key space [a-z].
    1956             :         //
    1957             :         // L0
    1958             :         //     ↦ 000100* ↤       ↦   000102  ↤       ↦  000101*  ↤
    1959             :         // L1
    1960             :         //     ↦ 000004* ↤
    1961             :         //     a b c d e f g h i j k l m n o p q r s t u v w x y z
    1962             :         //
    1963             :         // * - currently compacting
    1964           1 :         if pc.outputLevel != nil && pc.outputLevel.level != 0 {
    1965           1 :                 for _, c := range env.inProgressCompactions {
    1966           1 :                         if pc.outputLevel.level != c.outputLevel {
    1967           1 :                                 continue
    1968             :                         }
    1969           1 :                         if base.InternalCompare(pc.cmp, c.largest, pc.smallest) < 0 ||
    1970           1 :                                 base.InternalCompare(pc.cmp, c.smallest, pc.largest) > 0 {
    1971           1 :                                 continue
    1972             :                         }
    1973             : 
    1974             :                         // The picked compaction and the in-progress compaction c are
    1975             :                         // outputting to the same region of the key space of the same
    1976             :                         // level.
    1977           1 :                         return true
    1978             :                 }
    1979             :         }
    1980           1 :         return false
    1981             : }
    1982             : 
    1983             : // conflictsWithInProgress checks if there are any in-progress compactions with overlapping keyspace.
    1984             : func conflictsWithInProgress(
    1985             :         manual *manualCompaction, outputLevel int, inProgressCompactions []compactionInfo, cmp Compare,
    1986           1 : ) bool {
    1987           1 :         for _, c := range inProgressCompactions {
    1988           1 :                 if (c.outputLevel == manual.level || c.outputLevel == outputLevel) &&
    1989           1 :                         isUserKeysOverlapping(manual.start, manual.end, c.smallest.UserKey, c.largest.UserKey, cmp) {
    1990           1 :                         return true
    1991           1 :                 }
    1992           1 :                 for _, in := range c.inputs {
    1993           1 :                         if in.files.Empty() {
    1994           1 :                                 continue
    1995             :                         }
    1996           1 :                         iter := in.files.Iter()
    1997           1 :                         smallest := iter.First().Smallest.UserKey
    1998           1 :                         largest := iter.Last().Largest.UserKey
    1999           1 :                         if (in.level == manual.level || in.level == outputLevel) &&
    2000           1 :                                 isUserKeysOverlapping(manual.start, manual.end, smallest, largest, cmp) {
    2001           1 :                                 return true
    2002           1 :                         }
    2003             :                 }
    2004             :         }
    2005           1 :         return false
    2006             : }
    2007             : 
    2008           1 : func isUserKeysOverlapping(x1, x2, y1, y2 []byte, cmp Compare) bool {
    2009           1 :         return cmp(x1, y2) <= 0 && cmp(y1, x2) <= 0
    2010           1 : }

Generated by: LCOV version 1.14