LCOV - code coverage report
Current view: top level - pebble - compaction.go (source / functions) Coverage Total Hit
Test: 2025-06-25 08:18Z 9f09afa9 - tests + meta.lcov Lines: 93.3 % 2486 2320
Test Date: 2025-06-25 08:20:02 Functions: - 0 0

            Line data    Source code
       1              : // Copyright 2013 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              :         "context"
      10              :         "fmt"
      11              :         "iter"
      12              :         "math"
      13              :         "runtime/pprof"
      14              :         "slices"
      15              :         "sort"
      16              :         "sync/atomic"
      17              :         "time"
      18              :         "unsafe"
      19              : 
      20              :         "github.com/cockroachdb/crlib/crtime"
      21              :         "github.com/cockroachdb/errors"
      22              :         "github.com/cockroachdb/pebble/internal/base"
      23              :         "github.com/cockroachdb/pebble/internal/compact"
      24              :         "github.com/cockroachdb/pebble/internal/keyspan"
      25              :         "github.com/cockroachdb/pebble/internal/keyspan/keyspanimpl"
      26              :         "github.com/cockroachdb/pebble/internal/manifest"
      27              :         "github.com/cockroachdb/pebble/internal/problemspans"
      28              :         "github.com/cockroachdb/pebble/internal/sstableinternal"
      29              :         "github.com/cockroachdb/pebble/objstorage"
      30              :         "github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
      31              :         "github.com/cockroachdb/pebble/objstorage/remote"
      32              :         "github.com/cockroachdb/pebble/sstable"
      33              :         "github.com/cockroachdb/pebble/sstable/blob"
      34              :         "github.com/cockroachdb/pebble/sstable/block"
      35              :         "github.com/cockroachdb/pebble/vfs"
      36              :         "github.com/cockroachdb/redact"
      37              : )
      38              : 
      39              : var errEmptyTable = errors.New("pebble: empty table")
      40              : 
      41              : // ErrCancelledCompaction is returned if a compaction is cancelled by a
      42              : // concurrent excise or ingest-split operation.
      43              : var ErrCancelledCompaction = errors.New("pebble: compaction cancelled by a concurrent operation, will retry compaction")
      44              : 
      45              : var flushLabels = pprof.Labels("pebble", "flush", "output-level", "L0")
      46              : var gcLabels = pprof.Labels("pebble", "gc")
      47              : 
      48              : // expandedCompactionByteSizeLimit is the maximum number of bytes in all
      49              : // compacted files. We avoid expanding the lower level file set of a compaction
      50              : // if it would make the total compaction cover more than this many bytes.
      51              : func expandedCompactionByteSizeLimit(
      52              :         opts *Options, targetFileSize int64, availBytes uint64,
      53            2 : ) uint64 {
      54            2 :         v := uint64(25 * targetFileSize)
      55            2 : 
      56            2 :         // Never expand a compaction beyond half the available capacity, divided
      57            2 :         // by the maximum number of concurrent compactions. Each of the concurrent
      58            2 :         // compactions may expand up to this limit, so this attempts to limit
      59            2 :         // compactions to half of available disk space. Note that this will not
      60            2 :         // prevent compaction picking from pursuing compactions that are larger
      61            2 :         // than this threshold before expansion.
      62            2 :         //
      63            2 :         // NB: this heuristic is an approximation since we may run more compactions
      64            2 :         // than the upper concurrency limit.
      65            2 :         _, maxConcurrency := opts.CompactionConcurrencyRange()
      66            2 :         diskMax := (availBytes / 2) / uint64(maxConcurrency)
      67            2 :         if v > diskMax {
      68            1 :                 v = diskMax
      69            1 :         }
      70            2 :         return v
      71              : }
      72              : 
      73              : // maxGrandparentOverlapBytes is the maximum bytes of overlap with level+1
      74              : // before we stop building a single file in a level-1 to level compaction.
      75            2 : func maxGrandparentOverlapBytes(targetFileSize int64) uint64 {
      76            2 :         return uint64(10 * targetFileSize)
      77            2 : }
      78              : 
      79              : // maxReadCompactionBytes is used to prevent read compactions which
      80              : // are too wide.
      81            2 : func maxReadCompactionBytes(targetFileSize int64) uint64 {
      82            2 :         return uint64(10 * targetFileSize)
      83            2 : }
      84              : 
      85              : // noCloseIter wraps around a FragmentIterator, intercepting and eliding
      86              : // calls to Close. It is used during compaction to ensure that rangeDelIters
      87              : // are not closed prematurely.
      88              : type noCloseIter struct {
      89              :         keyspan.FragmentIterator
      90              : }
      91              : 
      92            2 : func (i *noCloseIter) Close() {}
      93              : 
      94              : type compactionLevel struct {
      95              :         level int
      96              :         files manifest.LevelSlice
      97              :         // l0SublevelInfo contains information about L0 sublevels being compacted.
      98              :         // It's only set for the start level of a compaction starting out of L0 and
      99              :         // is nil for all other compactions.
     100              :         l0SublevelInfo []sublevelInfo
     101              : }
     102              : 
     103            2 : func (cl compactionLevel) Clone() compactionLevel {
     104            2 :         newCL := compactionLevel{
     105            2 :                 level: cl.level,
     106            2 :                 files: cl.files,
     107            2 :         }
     108            2 :         return newCL
     109            2 : }
     110            1 : func (cl compactionLevel) String() string {
     111            1 :         return fmt.Sprintf(`Level %d, Files %s`, cl.level, cl.files)
     112            1 : }
     113              : 
     114              : // compactionWritable is a objstorage.Writable wrapper that, on every write,
     115              : // updates a metric in `versions` on bytes written by in-progress compactions so
     116              : // far. It also increments a per-compaction `written` atomic int.
     117              : type compactionWritable struct {
     118              :         objstorage.Writable
     119              : 
     120              :         versions *versionSet
     121              :         written  *atomic.Int64
     122              : }
     123              : 
     124              : // Write is part of the objstorage.Writable interface.
     125            2 : func (c *compactionWritable) Write(p []byte) error {
     126            2 :         if err := c.Writable.Write(p); err != nil {
     127            0 :                 return err
     128            0 :         }
     129              : 
     130            2 :         c.written.Add(int64(len(p)))
     131            2 :         c.versions.incrementCompactionBytes(int64(len(p)))
     132            2 :         return nil
     133              : }
     134              : 
     135              : type compactionKind int
     136              : 
     137              : const (
     138              :         compactionKindDefault compactionKind = iota
     139              :         compactionKindFlush
     140              :         // compactionKindMove denotes a move compaction where the input file is
     141              :         // retained and linked in a new level without being obsoleted.
     142              :         compactionKindMove
     143              :         // compactionKindCopy denotes a copy compaction where the input file is
     144              :         // copied byte-by-byte into a new file with a new TableNum in the output level.
     145              :         compactionKindCopy
     146              :         // compactionKindDeleteOnly denotes a compaction that only deletes input
     147              :         // files. It can occur when wide range tombstones completely contain sstables.
     148              :         compactionKindDeleteOnly
     149              :         compactionKindElisionOnly
     150              :         compactionKindRead
     151              :         compactionKindTombstoneDensity
     152              :         compactionKindRewrite
     153              :         compactionKindIngestedFlushable
     154              :         compactionKindBlobFileRewrite
     155              : )
     156              : 
     157            2 : func (k compactionKind) String() string {
     158            2 :         switch k {
     159            2 :         case compactionKindDefault:
     160            2 :                 return "default"
     161            0 :         case compactionKindFlush:
     162            0 :                 return "flush"
     163            2 :         case compactionKindMove:
     164            2 :                 return "move"
     165            2 :         case compactionKindDeleteOnly:
     166            2 :                 return "delete-only"
     167            2 :         case compactionKindElisionOnly:
     168            2 :                 return "elision-only"
     169            1 :         case compactionKindRead:
     170            1 :                 return "read"
     171            2 :         case compactionKindTombstoneDensity:
     172            2 :                 return "tombstone-density"
     173            2 :         case compactionKindRewrite:
     174            2 :                 return "rewrite"
     175            0 :         case compactionKindIngestedFlushable:
     176            0 :                 return "ingested-flushable"
     177            2 :         case compactionKindCopy:
     178            2 :                 return "copy"
     179            0 :         case compactionKindBlobFileRewrite:
     180            0 :                 return "blob-file-rewrite"
     181              :         }
     182            0 :         return "?"
     183              : }
     184              : 
     185              : // compactingOrFlushing returns "flushing" if the compaction kind is a flush,
     186              : // otherwise it returns "compacting".
     187            2 : func (k compactionKind) compactingOrFlushing() string {
     188            2 :         if k == compactionKindFlush {
     189            2 :                 return "flushing"
     190            2 :         }
     191            2 :         return "compacting"
     192              : }
     193              : 
     194              : type compaction interface {
     195              :         AddInProgressLocked(*DB)
     196              :         BeganAt() time.Time
     197              :         Bounds() *base.UserKeyBounds
     198              :         Cancel()
     199              :         Execute(JobID, *DB) error
     200              :         GrantHandle() CompactionGrantHandle
     201              :         Info() compactionInfo
     202              :         IsDownload() bool
     203              :         IsFlush() bool
     204              :         PprofLabels(UserKeyCategories) pprof.LabelSet
     205              :         RecordError(*problemspans.ByLevel, error)
     206              :         Tables() iter.Seq2[int, *manifest.TableMetadata]
     207              :         VersionEditApplied() bool
     208              : }
     209              : 
     210              : // tableCompaction is a table compaction from one level to the next, starting
     211              : // from a given version. It implements the compaction interface.
     212              : type tableCompaction struct {
     213              :         // cancel is a bool that can be used by other goroutines to signal a compaction
     214              :         // to cancel, such as if a conflicting excise operation raced it to manifest
     215              :         // application. Only holders of the manifest lock will write to this atomic.
     216              :         cancel atomic.Bool
     217              :         // kind indicates the kind of compaction. Different compaction kinds have
     218              :         // different semantics and mechanics. Some may have additional fields.
     219              :         kind compactionKind
     220              :         // isDownload is true if this compaction was started as part of a Download
     221              :         // operation. In this case kind is compactionKindCopy or
     222              :         // compactionKindRewrite.
     223              :         isDownload bool
     224              : 
     225              :         comparer *base.Comparer
     226              :         logger   Logger
     227              :         version  *manifest.Version
     228              :         // versionEditApplied is set to true when a compaction has completed and the
     229              :         // resulting version has been installed (if successful), but the compaction
     230              :         // goroutine is still cleaning up (eg, deleting obsolete files).
     231              :         versionEditApplied bool
     232              :         // getValueSeparation constructs a compact.ValueSeparation for use in a
     233              :         // compaction. It implements heuristics around choosing whether a compaction
     234              :         // should:
     235              :         //
     236              :         // a) preserve existing blob references: The compaction does not write any
     237              :         // new blob files, but propagates existing references to blob files.This
     238              :         // conserves write bandwidth by avoiding rewriting the referenced values. It
     239              :         // also reduces the locality of the referenced values which can reduce scan
     240              :         // performance because a scan must load values from more unique blob files.
     241              :         // It can also delay reclamation of disk space if some of the references to
     242              :         // blob values are elided by the compaction, increasing space amplification.
     243              :         //
     244              :         // b) rewrite blob files: The compaction will write eligible values to new
     245              :         // blob files. This consumes more write bandwidth because all values are
     246              :         // rewritten. However it restores locality.
     247              :         getValueSeparation func(JobID, *tableCompaction, sstable.TableFormat) compact.ValueSeparation
     248              : 
     249              :         // startLevel is the level that is being compacted. Inputs from startLevel
     250              :         // and outputLevel will be merged to produce a set of outputLevel files.
     251              :         startLevel *compactionLevel
     252              : 
     253              :         // outputLevel is the level that files are being produced in. outputLevel is
     254              :         // equal to startLevel+1 except when:
     255              :         //    - if startLevel is 0, the output level equals compactionPicker.baseLevel().
     256              :         //    - in multilevel compaction, the output level is the lowest level involved in
     257              :         //      the compaction
     258              :         // A compaction's outputLevel is nil for delete-only compactions.
     259              :         outputLevel *compactionLevel
     260              : 
     261              :         // extraLevels point to additional levels in between the input and output
     262              :         // levels that get compacted in multilevel compactions
     263              :         extraLevels []*compactionLevel
     264              : 
     265              :         inputs []compactionLevel
     266              : 
     267              :         // maxOutputFileSize is the maximum size of an individual table created
     268              :         // during compaction.
     269              :         maxOutputFileSize uint64
     270              :         // maxOverlapBytes is the maximum number of bytes of overlap allowed for a
     271              :         // single output table with the tables in the grandparent level.
     272              :         maxOverlapBytes uint64
     273              : 
     274              :         // The boundaries of the input data.
     275              :         bounds base.UserKeyBounds
     276              : 
     277              :         // grandparents are the tables in level+2 that overlap with the files being
     278              :         // compacted. Used to determine output table boundaries. Do not assume that the actual files
     279              :         // in the grandparent when this compaction finishes will be the same.
     280              :         grandparents manifest.LevelSlice
     281              : 
     282              :         delElision      compact.TombstoneElision
     283              :         rangeKeyElision compact.TombstoneElision
     284              : 
     285              :         // deleteOnly contains information specific to compactions with kind
     286              :         // compactionKindDeleteOnly. A delete-only compaction is a special
     287              :         // compaction that does not merge or write sstables. Instead, it only
     288              :         // performs deletions either through removing whole sstables from the LSM or
     289              :         // virtualizing them into virtual sstables.
     290              :         deleteOnly struct {
     291              :                 // hints are collected by the table stats collector and describe range
     292              :                 // deletions and the files containing keys deleted by them.
     293              :                 hints []deleteCompactionHint
     294              :                 // exciseEnabled is set to true if this compaction is allowed to excise
     295              :                 // files. If false, the compaction will only remove whole sstables that
     296              :                 // are wholly contained within the bounds of range deletions.
     297              :                 exciseEnabled bool
     298              :         }
     299              :         // flush contains information specific to flushes (compactionKindFlush and
     300              :         // compactionKindIngestedFlushable). A flush is modeled by a compaction
     301              :         // because it has similar mechanics to a default compaction.
     302              :         flush struct {
     303              :                 // flushables contains the flushables (aka memtables, large batches,
     304              :                 // flushable ingestions, etc) that are being flushed.
     305              :                 flushables flushableList
     306              :                 // Boundaries at which sstables flushed to L0 should be split.
     307              :                 // Determined by L0Sublevels. If nil, ignored.
     308              :                 l0Limits [][]byte
     309              :         }
     310              :         // iterationState contains state used during compaction iteration.
     311              :         iterationState struct {
     312              :                 // bufferPool is a pool of buffers used when reading blocks. Compactions
     313              :                 // do not populate the block cache under the assumption that the blocks
     314              :                 // we read will soon be irrelevant when their containing sstables are
     315              :                 // removed from the LSM.
     316              :                 bufferPool sstable.BufferPool
     317              :                 // keyspanIterClosers is a list of fragment iterators to close when the
     318              :                 // compaction finishes. As iteration opens new keyspan iterators,
     319              :                 // elements are appended. Keyspan iterators must remain open for the
     320              :                 // lifetime of the compaction, so they're accumulated here. When the
     321              :                 // compaction finishes, all the underlying keyspan iterators are closed.
     322              :                 keyspanIterClosers []*noCloseIter
     323              :                 // valueFetcher is used to fetch values from blob files. It's propagated
     324              :                 // down the iterator tree through the internal iterator options.
     325              :                 valueFetcher blob.ValueFetcher
     326              :         }
     327              :         // metrics encapsulates various metrics collected during a compaction.
     328              :         metrics compactionMetrics
     329              : 
     330              :         grantHandle CompactionGrantHandle
     331              : 
     332              :         tableFormat   sstable.TableFormat
     333              :         objCreateOpts objstorage.CreateOptions
     334              : }
     335              : 
     336              : // Assert that tableCompaction implements the compaction interface.
     337              : var _ compaction = (*tableCompaction)(nil)
     338              : 
     339            2 : func (c *tableCompaction) AddInProgressLocked(d *DB) {
     340            2 :         d.mu.compact.inProgress[c] = struct{}{}
     341            2 :         var isBase, isIntraL0 bool
     342            2 :         for _, cl := range c.inputs {
     343            2 :                 for f := range cl.files.All() {
     344            2 :                         if f.IsCompacting() {
     345            0 :                                 d.opts.Logger.Fatalf("L%d->L%d: %s already being compacted", c.startLevel.level, c.outputLevel.level, f.TableNum)
     346            0 :                         }
     347            2 :                         f.SetCompactionState(manifest.CompactionStateCompacting)
     348            2 :                         if c.startLevel != nil && c.outputLevel != nil && c.startLevel.level == 0 {
     349            2 :                                 if c.outputLevel.level == 0 {
     350            2 :                                         f.IsIntraL0Compacting = true
     351            2 :                                         isIntraL0 = true
     352            2 :                                 } else {
     353            2 :                                         isBase = true
     354            2 :                                 }
     355              :                         }
     356              :                 }
     357              :         }
     358              : 
     359            2 :         if isIntraL0 || isBase {
     360            2 :                 l0Inputs := []manifest.LevelSlice{c.startLevel.files}
     361            2 :                 if isIntraL0 {
     362            2 :                         l0Inputs = append(l0Inputs, c.outputLevel.files)
     363            2 :                 }
     364            2 :                 if err := d.mu.versions.latest.l0Organizer.UpdateStateForStartedCompaction(l0Inputs, isBase); err != nil {
     365            0 :                         d.opts.Logger.Fatalf("could not update state for compaction: %s", err)
     366            0 :                 }
     367              :         }
     368              : }
     369              : 
     370            2 : func (c *tableCompaction) BeganAt() time.Time          { return c.metrics.beganAt }
     371            2 : func (c *tableCompaction) Bounds() *base.UserKeyBounds { return &c.bounds }
     372            2 : func (c *tableCompaction) Cancel()                     { c.cancel.Store(true) }
     373              : 
     374            2 : func (c *tableCompaction) Execute(jobID JobID, d *DB) error {
     375            2 :         c.grantHandle.Started()
     376            2 :         err := d.compact1(jobID, c)
     377            2 :         // The version stored in the compaction is ref'd when the compaction is
     378            2 :         // created. We're responsible for un-refing it when the compaction is
     379            2 :         // complete.
     380            2 :         if c.version != nil {
     381            2 :                 c.version.UnrefLocked()
     382            2 :         }
     383            2 :         return err
     384              : }
     385              : 
     386            1 : func (c *tableCompaction) RecordError(problemSpans *problemspans.ByLevel, err error) {
     387            1 :         // Record problem spans for a short duration, unless the error is a
     388            1 :         // corruption.
     389            1 :         expiration := 30 * time.Second
     390            1 :         if IsCorruptionError(err) {
     391            1 :                 // TODO(radu): ideally, we should be using the corruption reporting
     392            1 :                 // mechanism which has a tighter span for the corruption. We would need to
     393            1 :                 // somehow plumb the level of the file.
     394            1 :                 expiration = 5 * time.Minute
     395            1 :         }
     396              : 
     397            1 :         for i := range c.inputs {
     398            1 :                 level := c.inputs[i].level
     399            1 :                 if level == 0 {
     400            1 :                         // We do not set problem spans on L0, as they could block flushes.
     401            1 :                         continue
     402              :                 }
     403            1 :                 it := c.inputs[i].files.Iter()
     404            1 :                 for f := it.First(); f != nil; f = it.Next() {
     405            1 :                         problemSpans.Add(level, f.UserKeyBounds(), expiration)
     406            1 :                 }
     407              :         }
     408              : }
     409              : 
     410            2 : func (c *tableCompaction) GrantHandle() CompactionGrantHandle { return c.grantHandle }
     411            2 : func (c *tableCompaction) IsDownload() bool                   { return c.isDownload }
     412            2 : func (c *tableCompaction) IsFlush() bool                      { return len(c.flush.flushables) > 0 }
     413            2 : func (c *tableCompaction) Info() compactionInfo {
     414            2 :         info := compactionInfo{
     415            2 :                 versionEditApplied: c.versionEditApplied,
     416            2 :                 kind:               c.kind,
     417            2 :                 inputs:             c.inputs,
     418            2 :                 bounds:             &c.bounds,
     419            2 :                 outputLevel:        -1,
     420            2 :         }
     421            2 :         if c.outputLevel != nil {
     422            2 :                 info.outputLevel = c.outputLevel.level
     423            2 :         }
     424            2 :         return info
     425              : }
     426            2 : func (c *tableCompaction) PprofLabels(kc UserKeyCategories) pprof.LabelSet {
     427            2 :         activity := "compact"
     428            2 :         if len(c.flush.flushables) != 0 {
     429            0 :                 activity = "flush"
     430            0 :         }
     431            2 :         level := "L?"
     432            2 :         // Delete-only compactions don't have an output level.
     433            2 :         if c.outputLevel != nil {
     434            2 :                 level = fmt.Sprintf("L%d", c.outputLevel.level)
     435            2 :         }
     436            2 :         if kc.Len() > 0 {
     437            0 :                 cat := kc.CategorizeKeyRange(c.bounds.Start, c.bounds.End.Key)
     438            0 :                 return pprof.Labels("pebble", activity, "output-level", level, "key-type", cat)
     439            0 :         }
     440            2 :         return pprof.Labels("pebble", activity, "output-level", level)
     441              : }
     442              : 
     443            2 : func (c *tableCompaction) Tables() iter.Seq2[int, *manifest.TableMetadata] {
     444            2 :         return func(yield func(int, *manifest.TableMetadata) bool) {
     445            2 :                 for _, cl := range c.inputs {
     446            2 :                         for f := range cl.files.All() {
     447            2 :                                 if !yield(cl.level, f) {
     448            2 :                                         return
     449            2 :                                 }
     450              :                         }
     451              :                 }
     452              :         }
     453              : }
     454              : 
     455            2 : func (c *tableCompaction) VersionEditApplied() bool { return c.versionEditApplied }
     456              : 
     457              : // compactionMetrics contians metrics surrounding a compaction.
     458              : type compactionMetrics struct {
     459              :         // beganAt is the time when the compaction began.
     460              :         beganAt time.Time
     461              :         // bytesWritten contains the number of bytes that have been written to
     462              :         // outputs. It's updated whenever the compaction outputs'
     463              :         // objstorage.Writables receive new writes. See newCompactionOutputObj.
     464              :         bytesWritten atomic.Int64
     465              :         // internalIterStats contains statistics from the internal iterators used by
     466              :         // the compaction.
     467              :         //
     468              :         // TODO(jackson): Use these to power the compaction BytesRead metric.
     469              :         internalIterStats base.InternalIteratorStats
     470              :         // perLevel contains metrics for each level involved in the compaction.
     471              :         perLevel levelMetricsDelta
     472              :         // picker contains metrics from the compaction picker when the compaction
     473              :         // was picked.
     474              :         picker pickedCompactionMetrics
     475              : }
     476              : 
     477              : // inputLargestSeqNumAbsolute returns the maximum LargestSeqNumAbsolute of any
     478              : // input sstables.
     479            2 : func (c *tableCompaction) inputLargestSeqNumAbsolute() base.SeqNum {
     480            2 :         var seqNum base.SeqNum
     481            2 :         for _, cl := range c.inputs {
     482            2 :                 for m := range cl.files.All() {
     483            2 :                         seqNum = max(seqNum, m.LargestSeqNumAbsolute)
     484            2 :                 }
     485              :         }
     486            2 :         return seqNum
     487              : }
     488              : 
     489            2 : func (c *tableCompaction) makeInfo(jobID JobID) CompactionInfo {
     490            2 :         info := CompactionInfo{
     491            2 :                 JobID:       int(jobID),
     492            2 :                 Reason:      c.kind.String(),
     493            2 :                 Input:       make([]LevelInfo, 0, len(c.inputs)),
     494            2 :                 Annotations: []string{},
     495            2 :         }
     496            2 :         if c.isDownload {
     497            2 :                 info.Reason = "download," + info.Reason
     498            2 :         }
     499            2 :         for _, cl := range c.inputs {
     500            2 :                 inputInfo := LevelInfo{Level: cl.level, Tables: nil}
     501            2 :                 for m := range cl.files.All() {
     502            2 :                         inputInfo.Tables = append(inputInfo.Tables, m.TableInfo())
     503            2 :                 }
     504            2 :                 info.Input = append(info.Input, inputInfo)
     505              :         }
     506            2 :         if c.outputLevel != nil {
     507            2 :                 info.Output.Level = c.outputLevel.level
     508            2 : 
     509            2 :                 // If there are no inputs from the output level (eg, a move
     510            2 :                 // compaction), add an empty LevelInfo to info.Input.
     511            2 :                 if len(c.inputs) > 0 && c.inputs[len(c.inputs)-1].level != c.outputLevel.level {
     512            0 :                         info.Input = append(info.Input, LevelInfo{Level: c.outputLevel.level})
     513            0 :                 }
     514            2 :         } else {
     515            2 :                 // For a delete-only compaction, set the output level to L6. The
     516            2 :                 // output level is not meaningful here, but complicating the
     517            2 :                 // info.Output interface with a pointer doesn't seem worth the
     518            2 :                 // semantic distinction.
     519            2 :                 info.Output.Level = numLevels - 1
     520            2 :         }
     521              : 
     522            2 :         for i, score := range c.metrics.picker.scores {
     523            2 :                 info.Input[i].Score = score
     524            2 :         }
     525            2 :         info.SingleLevelOverlappingRatio = c.metrics.picker.singleLevelOverlappingRatio
     526            2 :         info.MultiLevelOverlappingRatio = c.metrics.picker.multiLevelOverlappingRatio
     527            2 :         if len(info.Input) > 2 {
     528            2 :                 info.Annotations = append(info.Annotations, "multilevel")
     529            2 :         }
     530            2 :         return info
     531              : }
     532              : 
     533              : type getValueSeparation func(JobID, *tableCompaction, sstable.TableFormat) compact.ValueSeparation
     534              : 
     535              : // newCompaction constructs a compaction from the provided picked compaction.
     536              : //
     537              : // The compaction is created with a reference to its version that must be
     538              : // released when the compaction is complete.
     539              : func newCompaction(
     540              :         pc *pickedTableCompaction,
     541              :         opts *Options,
     542              :         beganAt time.Time,
     543              :         provider objstorage.Provider,
     544              :         grantHandle CompactionGrantHandle,
     545              :         tableFormat sstable.TableFormat,
     546              :         getValueSeparation getValueSeparation,
     547            2 : ) *tableCompaction {
     548            2 :         c := &tableCompaction{
     549            2 :                 kind:               compactionKindDefault,
     550            2 :                 comparer:           opts.Comparer,
     551            2 :                 inputs:             pc.inputs,
     552            2 :                 bounds:             pc.bounds,
     553            2 :                 logger:             opts.Logger,
     554            2 :                 version:            pc.version,
     555            2 :                 getValueSeparation: getValueSeparation,
     556            2 :                 maxOutputFileSize:  pc.maxOutputFileSize,
     557            2 :                 maxOverlapBytes:    pc.maxOverlapBytes,
     558            2 :                 metrics: compactionMetrics{
     559            2 :                         beganAt: beganAt,
     560            2 :                         picker:  pc.pickerMetrics,
     561            2 :                 },
     562            2 :                 grantHandle: grantHandle,
     563            2 :                 tableFormat: tableFormat,
     564            2 :         }
     565            2 :         // Acquire a reference to the version to ensure that files and in-memory
     566            2 :         // version state necessary for reading files remain available. Ignoring
     567            2 :         // excises, this isn't strictly necessary for reading the sstables that are
     568            2 :         // inputs to the compaction because those files are 'marked as compacting'
     569            2 :         // and shouldn't be subject to any competing compactions. However with
     570            2 :         // excises, a concurrent excise may remove a compaction's file from the
     571            2 :         // Version and then cancel the compaction. The file shouldn't be physically
     572            2 :         // removed until the cancelled compaction stops reading it.
     573            2 :         //
     574            2 :         // Additionally, we need any blob files referenced by input sstables to
     575            2 :         // remain available, even if the blob file is rewritten. Maintaining a
     576            2 :         // reference ensures that all these files remain available for the
     577            2 :         // compaction's reads.
     578            2 :         c.version.Ref()
     579            2 : 
     580            2 :         c.startLevel = &c.inputs[0]
     581            2 :         if pc.startLevel.l0SublevelInfo != nil {
     582            2 :                 c.startLevel.l0SublevelInfo = pc.startLevel.l0SublevelInfo
     583            2 :         }
     584              : 
     585            2 :         c.outputLevel = &c.inputs[len(c.inputs)-1]
     586            2 : 
     587            2 :         if len(pc.inputs) > 2 {
     588            2 :                 // TODO(xinhaoz): Look into removing extraLevels on the compaction struct.
     589            2 :                 c.extraLevels = make([]*compactionLevel, 0, len(pc.inputs)-2)
     590            2 :                 for i := 1; i < len(pc.inputs)-1; i++ {
     591            2 :                         c.extraLevels = append(c.extraLevels, &c.inputs[i])
     592            2 :                 }
     593              :         }
     594              :         // Compute the set of outputLevel+1 files that overlap this compaction (these
     595              :         // are the grandparent sstables).
     596            2 :         if c.outputLevel.level+1 < numLevels {
     597            2 :                 c.grandparents = c.version.Overlaps(c.outputLevel.level+1, c.bounds)
     598            2 :         }
     599            2 :         c.delElision, c.rangeKeyElision = compact.SetupTombstoneElision(
     600            2 :                 c.comparer.Compare, c.version, pc.l0Organizer, c.outputLevel.level, c.bounds,
     601            2 :         )
     602            2 :         c.kind = pc.kind
     603            2 : 
     604            2 :         preferSharedStorage := tableFormat >= FormatMinForSharedObjects.MaxTableFormat() &&
     605            2 :                 remote.ShouldCreateShared(opts.Experimental.CreateOnShared, c.outputLevel.level)
     606            2 :         c.maybeSwitchToMoveOrCopy(preferSharedStorage, provider)
     607            2 :         c.objCreateOpts = objstorage.CreateOptions{
     608            2 :                 PreferSharedStorage: preferSharedStorage,
     609            2 :                 WriteCategory:       getDiskWriteCategoryForCompaction(opts, c.kind),
     610            2 :         }
     611            2 :         if preferSharedStorage {
     612            2 :                 c.getValueSeparation = neverSeparateValues
     613            2 :         }
     614              : 
     615            2 :         return c
     616              : }
     617              : 
     618              : // maybeSwitchToMoveOrCopy decides if the compaction can be changed into a move
     619              : // or copy compaction, in which case c.kind is updated.
     620              : func (c *tableCompaction) maybeSwitchToMoveOrCopy(
     621              :         preferSharedStorage bool, provider objstorage.Provider,
     622            2 : ) {
     623            2 :         // Only non-multi-level compactions with a single input file can be
     624            2 :         // considered.
     625            2 :         if c.startLevel.files.Len() != 1 || !c.outputLevel.files.Empty() || c.hasExtraLevelData() {
     626            2 :                 return
     627            2 :         }
     628              : 
     629              :         // In addition to the default compaction, we also check whether a tombstone
     630              :         // density compaction can be optimized into a move compaction. However, we
     631              :         // want to avoid performing a move compaction into the lowest level, since the
     632              :         // goal there is to actually remove the tombstones.
     633              :         //
     634              :         // Tombstone density compaction is meant to address cases where tombstones
     635              :         // don't reclaim much space but are still expensive to scan over. We can only
     636              :         // remove the tombstones once there's nothing at all underneath them.
     637            2 :         switch c.kind {
     638            2 :         case compactionKindDefault:
     639              :                 // Proceed.
     640            2 :         case compactionKindTombstoneDensity:
     641            2 :                 // Tombstone density compaction can be optimized into a move compaction.
     642            2 :                 // However, we want to avoid performing a move compaction into the lowest
     643            2 :                 // level, since the goal there is to actually remove the tombstones; even if
     644            2 :                 // they don't prevent a lot of space from being reclaimed, tombstones can
     645            2 :                 // still be expensive to scan over.
     646            2 :                 if c.outputLevel.level == numLevels-1 {
     647            2 :                         return
     648            2 :                 }
     649            2 :         default:
     650            2 :                 // Other compaction kinds not supported.
     651            2 :                 return
     652              :         }
     653              : 
     654              :         // We avoid a move or copy if there is lots of overlapping grandparent data.
     655              :         // Otherwise, the move could create a parent file that will require a very
     656              :         // expensive merge later on.
     657            2 :         if c.grandparents.AggregateSizeSum() > c.maxOverlapBytes {
     658            1 :                 return
     659            1 :         }
     660              : 
     661            2 :         iter := c.startLevel.files.Iter()
     662            2 :         meta := iter.First()
     663            2 : 
     664            2 :         // We should always be passed a provider, except in some unit tests.
     665            2 :         isRemote := provider != nil && !objstorage.IsLocalTable(provider, meta.TableBacking.DiskFileNum)
     666            2 : 
     667            2 :         // Shared and external tables can always be moved. We can also move a local
     668            2 :         // table unless we need the result to be on shared storage.
     669            2 :         if isRemote || !preferSharedStorage {
     670            2 :                 c.kind = compactionKindMove
     671            2 :                 return
     672            2 :         }
     673              : 
     674              :         // We can rewrite the table (regular compaction) or we can use a copy compaction.
     675            2 :         switch {
     676            1 :         case meta.Virtual:
     677              :                 // We want to avoid a copy compaction if the table is virtual, as we may end
     678              :                 // up copying a lot more data than necessary.
     679            2 :         case meta.BlobReferenceDepth != 0:
     680              :                 // We also want to avoid copy compactions for tables with blob references,
     681              :                 // as we currently lack a mechanism to propagate blob references along with
     682              :                 // the sstable.
     683            2 :         default:
     684            2 :                 c.kind = compactionKindCopy
     685              :         }
     686              : }
     687              : 
     688              : // newDeleteOnlyCompaction constructs a delete-only compaction from the provided
     689              : // inputs.
     690              : //
     691              : // The compaction is created with a reference to its version that must be
     692              : // released when the compaction is complete.
     693              : func newDeleteOnlyCompaction(
     694              :         opts *Options,
     695              :         cur *manifest.Version,
     696              :         inputs []compactionLevel,
     697              :         beganAt time.Time,
     698              :         hints []deleteCompactionHint,
     699              :         exciseEnabled bool,
     700            2 : ) *tableCompaction {
     701            2 :         c := &tableCompaction{
     702            2 :                 kind:        compactionKindDeleteOnly,
     703            2 :                 comparer:    opts.Comparer,
     704            2 :                 logger:      opts.Logger,
     705            2 :                 version:     cur,
     706            2 :                 inputs:      inputs,
     707            2 :                 grantHandle: noopGrantHandle{},
     708            2 :                 metrics: compactionMetrics{
     709            2 :                         beganAt: beganAt,
     710            2 :                 },
     711            2 :         }
     712            2 :         c.deleteOnly.hints = hints
     713            2 :         c.deleteOnly.exciseEnabled = exciseEnabled
     714            2 :         // Acquire a reference to the version to ensure that files and in-memory
     715            2 :         // version state necessary for reading files remain available. Ignoring
     716            2 :         // excises, this isn't strictly necessary for reading the sstables that are
     717            2 :         // inputs to the compaction because those files are 'marked as compacting'
     718            2 :         // and shouldn't be subject to any competing compactions. However with
     719            2 :         // excises, a concurrent excise may remove a compaction's file from the
     720            2 :         // Version and then cancel the compaction. The file shouldn't be physically
     721            2 :         // removed until the cancelled compaction stops reading it.
     722            2 :         //
     723            2 :         // Additionally, we need any blob files referenced by input sstables to
     724            2 :         // remain available, even if the blob file is rewritten. Maintaining a
     725            2 :         // reference ensures that all these files remain available for the
     726            2 :         // compaction's reads.
     727            2 :         c.version.Ref()
     728            2 : 
     729            2 :         // Set c.smallest, c.largest.
     730            2 :         cmp := opts.Comparer.Compare
     731            2 :         for _, in := range inputs {
     732            2 :                 c.bounds = manifest.ExtendKeyRange(cmp, c.bounds, in.files.All())
     733            2 :         }
     734            2 :         return c
     735              : }
     736              : 
     737            2 : func adjustGrandparentOverlapBytesForFlush(c *tableCompaction, flushingBytes uint64) {
     738            2 :         // Heuristic to place a lower bound on compaction output file size
     739            2 :         // caused by Lbase. Prior to this heuristic we have observed an L0 in
     740            2 :         // production with 310K files of which 290K files were < 10KB in size.
     741            2 :         // Our hypothesis is that it was caused by L1 having 2600 files and
     742            2 :         // ~10GB, such that each flush got split into many tiny files due to
     743            2 :         // overlapping with most of the files in Lbase.
     744            2 :         //
     745            2 :         // The computation below is general in that it accounts
     746            2 :         // for flushing different volumes of data (e.g. we may be flushing
     747            2 :         // many memtables). For illustration, we consider the typical
     748            2 :         // example of flushing a 64MB memtable. So 12.8MB output,
     749            2 :         // based on the compression guess below. If the compressed bytes
     750            2 :         // guess is an over-estimate we will end up with smaller files,
     751            2 :         // and if an under-estimate we will end up with larger files.
     752            2 :         // With a 2MB target file size, 7 files. We are willing to accept
     753            2 :         // 4x the number of files, if it results in better write amplification
     754            2 :         // when later compacting to Lbase, i.e., ~450KB files (target file
     755            2 :         // size / 4).
     756            2 :         //
     757            2 :         // Note that this is a pessimistic heuristic in that
     758            2 :         // fileCountUpperBoundDueToGrandparents could be far from the actual
     759            2 :         // number of files produced due to the grandparent limits. For
     760            2 :         // example, in the extreme, consider a flush that overlaps with 1000
     761            2 :         // files in Lbase f0...f999, and the initially calculated value of
     762            2 :         // maxOverlapBytes will cause splits at f10, f20,..., f990, which
     763            2 :         // means an upper bound file count of 100 files. Say the input bytes
     764            2 :         // in the flush are such that acceptableFileCount=10. We will fatten
     765            2 :         // up maxOverlapBytes by 10x to ensure that the upper bound file count
     766            2 :         // drops to 10. However, it is possible that in practice, even without
     767            2 :         // this change, we would have produced no more than 10 files, and that
     768            2 :         // this change makes the files unnecessarily wide. Say the input bytes
     769            2 :         // are distributed such that 10% are in f0...f9, 10% in f10...f19, ...
     770            2 :         // 10% in f80...f89 and 10% in f990...f999. The original value of
     771            2 :         // maxOverlapBytes would have actually produced only 10 sstables. But
     772            2 :         // by increasing maxOverlapBytes by 10x, we may produce 1 sstable that
     773            2 :         // spans f0...f89, i.e., a much wider sstable than necessary.
     774            2 :         //
     775            2 :         // We could produce a tighter estimate of
     776            2 :         // fileCountUpperBoundDueToGrandparents if we had knowledge of the key
     777            2 :         // distribution of the flush. The 4x multiplier mentioned earlier is
     778            2 :         // a way to try to compensate for this pessimism.
     779            2 :         //
     780            2 :         // TODO(sumeer): we don't have compression info for the data being
     781            2 :         // flushed, but it is likely that existing files that overlap with
     782            2 :         // this flush in Lbase are representative wrt compression ratio. We
     783            2 :         // could store the uncompressed size in TableMetadata and estimate
     784            2 :         // the compression ratio.
     785            2 :         const approxCompressionRatio = 0.2
     786            2 :         approxOutputBytes := approxCompressionRatio * float64(flushingBytes)
     787            2 :         approxNumFilesBasedOnTargetSize :=
     788            2 :                 int(math.Ceil(approxOutputBytes / float64(c.maxOutputFileSize)))
     789            2 :         acceptableFileCount := float64(4 * approxNumFilesBasedOnTargetSize)
     790            2 :         // The byte calculation is linear in numGrandparentFiles, but we will
     791            2 :         // incur this linear cost in compact.Runner.TableSplitLimit() too, so we are
     792            2 :         // also willing to pay it now. We could approximate this cheaply by using the
     793            2 :         // mean file size of Lbase.
     794            2 :         grandparentFileBytes := c.grandparents.AggregateSizeSum()
     795            2 :         fileCountUpperBoundDueToGrandparents :=
     796            2 :                 float64(grandparentFileBytes) / float64(c.maxOverlapBytes)
     797            2 :         if fileCountUpperBoundDueToGrandparents > acceptableFileCount {
     798            2 :                 c.maxOverlapBytes = uint64(
     799            2 :                         float64(c.maxOverlapBytes) *
     800            2 :                                 (fileCountUpperBoundDueToGrandparents / acceptableFileCount))
     801            2 :         }
     802              : }
     803              : 
     804              : // newFlush creates the state necessary for a flush (modeled with the compaction
     805              : // struct).
     806              : //
     807              : // newFlush takes the current Version in order to populate grandparent flushing
     808              : // limits, but it does not reference the version.
     809              : //
     810              : // TODO(jackson): Consider maintaining a reference to the version anyways since
     811              : // in the future in-memory Version state may only be available while a Version
     812              : // is referenced (eg, if we start recycling B-Tree nodes once they're no longer
     813              : // referenced). There's subtlety around unref'ing the version at the right
     814              : // moment, so we defer it for now.
     815              : func newFlush(
     816              :         opts *Options,
     817              :         cur *manifest.Version,
     818              :         l0Organizer *manifest.L0Organizer,
     819              :         baseLevel int,
     820              :         flushing flushableList,
     821              :         beganAt time.Time,
     822              :         tableFormat sstable.TableFormat,
     823              :         getValueSeparation getValueSeparation,
     824            2 : ) (*tableCompaction, error) {
     825            2 :         c := &tableCompaction{
     826            2 :                 kind:               compactionKindFlush,
     827            2 :                 comparer:           opts.Comparer,
     828            2 :                 logger:             opts.Logger,
     829            2 :                 inputs:             []compactionLevel{{level: -1}, {level: 0}},
     830            2 :                 getValueSeparation: getValueSeparation,
     831            2 :                 maxOutputFileSize:  math.MaxUint64,
     832            2 :                 maxOverlapBytes:    math.MaxUint64,
     833            2 :                 grantHandle:        noopGrantHandle{},
     834            2 :                 tableFormat:        tableFormat,
     835            2 :                 metrics: compactionMetrics{
     836            2 :                         beganAt: beganAt,
     837            2 :                 },
     838            2 :         }
     839            2 :         c.flush.flushables = flushing
     840            2 :         c.flush.l0Limits = l0Organizer.FlushSplitKeys()
     841            2 :         c.startLevel = &c.inputs[0]
     842            2 :         c.outputLevel = &c.inputs[1]
     843            2 :         if len(flushing) > 0 {
     844            2 :                 if _, ok := flushing[0].flushable.(*ingestedFlushable); ok {
     845            2 :                         if len(flushing) != 1 {
     846            0 :                                 panic("pebble: ingestedFlushable must be flushed one at a time.")
     847              :                         }
     848            2 :                         c.kind = compactionKindIngestedFlushable
     849            2 :                         return c, nil
     850            2 :                 } else {
     851            2 :                         // Make sure there's no ingestedFlushable after the first flushable
     852            2 :                         // in the list.
     853            2 :                         for _, f := range c.flush.flushables[1:] {
     854            2 :                                 if _, ok := f.flushable.(*ingestedFlushable); ok {
     855            0 :                                         panic("pebble: flushables shouldn't contain ingestedFlushable")
     856              :                                 }
     857              :                         }
     858              :                 }
     859              :         }
     860              : 
     861            2 :         preferSharedStorage := tableFormat >= FormatMinForSharedObjects.MaxTableFormat() &&
     862            2 :                 remote.ShouldCreateShared(opts.Experimental.CreateOnShared, c.outputLevel.level)
     863            2 :         c.objCreateOpts = objstorage.CreateOptions{
     864            2 :                 PreferSharedStorage: preferSharedStorage,
     865            2 :                 WriteCategory:       getDiskWriteCategoryForCompaction(opts, c.kind),
     866            2 :         }
     867            2 :         if preferSharedStorage {
     868            2 :                 c.getValueSeparation = neverSeparateValues
     869            2 :         }
     870              : 
     871            2 :         cmp := c.comparer.Compare
     872            2 :         updatePointBounds := func(iter internalIterator) {
     873            2 :                 if kv := iter.First(); kv != nil {
     874            2 :                         if c.bounds.Start == nil || cmp(c.bounds.Start, kv.K.UserKey) > 0 {
     875            2 :                                 c.bounds.Start = slices.Clone(kv.K.UserKey)
     876            2 :                         }
     877              :                 }
     878            2 :                 if kv := iter.Last(); kv != nil {
     879            2 :                         if c.bounds.End.Key == nil || !c.bounds.End.IsUpperBoundForInternalKey(cmp, kv.K) {
     880            2 :                                 c.bounds.End = base.UserKeyExclusiveIf(slices.Clone(kv.K.UserKey), kv.K.IsExclusiveSentinel())
     881            2 :                         }
     882              :                 }
     883              :         }
     884              : 
     885            2 :         updateRangeBounds := func(iter keyspan.FragmentIterator) error {
     886            2 :                 // File bounds require s != nil && !s.Empty(). We only need to check for
     887            2 :                 // s != nil here, as the memtable's FragmentIterator would never surface
     888            2 :                 // empty spans.
     889            2 :                 if s, err := iter.First(); err != nil {
     890            0 :                         return err
     891            2 :                 } else if s != nil {
     892            2 :                         c.bounds = c.bounds.Union(cmp, s.Bounds().Clone())
     893            2 :                 }
     894            2 :                 if s, err := iter.Last(); err != nil {
     895            0 :                         return err
     896            2 :                 } else if s != nil {
     897            2 :                         c.bounds = c.bounds.Union(cmp, s.Bounds().Clone())
     898            2 :                 }
     899            2 :                 return nil
     900              :         }
     901              : 
     902            2 :         var flushingBytes uint64
     903            2 :         for i := range flushing {
     904            2 :                 f := flushing[i]
     905            2 :                 updatePointBounds(f.newIter(nil))
     906            2 :                 if rangeDelIter := f.newRangeDelIter(nil); rangeDelIter != nil {
     907            2 :                         if err := updateRangeBounds(rangeDelIter); err != nil {
     908            0 :                                 return nil, err
     909            0 :                         }
     910              :                 }
     911            2 :                 if rangeKeyIter := f.newRangeKeyIter(nil); rangeKeyIter != nil {
     912            2 :                         if err := updateRangeBounds(rangeKeyIter); err != nil {
     913            0 :                                 return nil, err
     914            0 :                         }
     915              :                 }
     916            2 :                 flushingBytes += f.inuseBytes()
     917              :         }
     918              : 
     919            2 :         if opts.FlushSplitBytes > 0 {
     920            2 :                 c.maxOutputFileSize = uint64(opts.TargetFileSizes[0])
     921            2 :                 c.maxOverlapBytes = maxGrandparentOverlapBytes(opts.TargetFileSizes[0])
     922            2 :                 c.grandparents = cur.Overlaps(baseLevel, c.bounds)
     923            2 :                 adjustGrandparentOverlapBytesForFlush(c, flushingBytes)
     924            2 :         }
     925              : 
     926              :         // We don't elide tombstones for flushes.
     927            2 :         c.delElision, c.rangeKeyElision = compact.NoTombstoneElision(), compact.NoTombstoneElision()
     928            2 :         return c, nil
     929              : }
     930              : 
     931            2 : func (c *tableCompaction) hasExtraLevelData() bool {
     932            2 :         if len(c.extraLevels) == 0 {
     933            2 :                 // not a multi level compaction
     934            2 :                 return false
     935            2 :         } else if c.extraLevels[0].files.Empty() {
     936            2 :                 // a multi level compaction without data in the intermediate input level;
     937            2 :                 // e.g. for a multi level compaction with levels 4,5, and 6, this could
     938            2 :                 // occur if there is no files to compact in 5, or in 5 and 6 (i.e. a move).
     939            2 :                 return false
     940            2 :         }
     941            2 :         return true
     942              : }
     943              : 
     944              : // errorOnUserKeyOverlap returns an error if the last two written sstables in
     945              : // this compaction have revisions of the same user key present in both sstables,
     946              : // when it shouldn't (eg. when splitting flushes).
     947            1 : func (c *tableCompaction) errorOnUserKeyOverlap(ve *manifest.VersionEdit) error {
     948            1 :         if n := len(ve.NewTables); n > 1 {
     949            1 :                 meta := ve.NewTables[n-1].Meta
     950            1 :                 prevMeta := ve.NewTables[n-2].Meta
     951            1 :                 if !prevMeta.Largest().IsExclusiveSentinel() &&
     952            1 :                         c.comparer.Compare(prevMeta.Largest().UserKey, meta.Smallest().UserKey) >= 0 {
     953            1 :                         return errors.Errorf("pebble: compaction split user key across two sstables: %s in %s and %s",
     954            1 :                                 prevMeta.Largest().Pretty(c.comparer.FormatKey),
     955            1 :                                 prevMeta.TableNum,
     956            1 :                                 meta.TableNum)
     957            1 :                 }
     958              :         }
     959            1 :         return nil
     960              : }
     961              : 
     962              : // allowZeroSeqNum returns true if seqnum's can be zeroed if there are no
     963              : // snapshots requiring them to be kept. It performs this determination by
     964              : // looking at the TombstoneElision values which are set up based on sstables
     965              : // which overlap the bounds of the compaction at a lower level in the LSM.
     966            2 : func (c *tableCompaction) allowZeroSeqNum() bool {
     967            2 :         // TODO(peter): we disable zeroing of seqnums during flushing to match
     968            2 :         // RocksDB behavior and to avoid generating overlapping sstables during
     969            2 :         // DB.replayWAL. When replaying WAL files at startup, we flush after each
     970            2 :         // WAL is replayed building up a single version edit that is
     971            2 :         // applied. Because we don't apply the version edit after each flush, this
     972            2 :         // code doesn't know that L0 contains files and zeroing of seqnums should
     973            2 :         // be disabled. That is fixable, but it seems safer to just match the
     974            2 :         // RocksDB behavior for now.
     975            2 :         return len(c.flush.flushables) == 0 && c.delElision.ElidesEverything() && c.rangeKeyElision.ElidesEverything()
     976            2 : }
     977              : 
     978              : // newInputIters returns an iterator over all the input tables in a compaction.
     979              : func (c *tableCompaction) newInputIters(
     980              :         newIters tableNewIters, iiopts internalIterOpts,
     981              : ) (
     982              :         pointIter internalIterator,
     983              :         rangeDelIter, rangeKeyIter keyspan.FragmentIterator,
     984              :         retErr error,
     985            2 : ) {
     986            2 :         ctx := context.TODO()
     987            2 :         cmp := c.comparer.Compare
     988            2 : 
     989            2 :         // Validate the ordering of compaction input files for defense in depth.
     990            2 :         if len(c.flush.flushables) == 0 {
     991            2 :                 if c.startLevel.level >= 0 {
     992            2 :                         err := manifest.CheckOrdering(c.comparer, manifest.Level(c.startLevel.level),
     993            2 :                                 c.startLevel.files.Iter())
     994            2 :                         if err != nil {
     995            1 :                                 return nil, nil, nil, err
     996            1 :                         }
     997              :                 }
     998            2 :                 err := manifest.CheckOrdering(c.comparer, manifest.Level(c.outputLevel.level),
     999            2 :                         c.outputLevel.files.Iter())
    1000            2 :                 if err != nil {
    1001            1 :                         return nil, nil, nil, err
    1002            1 :                 }
    1003            2 :                 if c.startLevel.level == 0 {
    1004            2 :                         if c.startLevel.l0SublevelInfo == nil {
    1005            0 :                                 panic("l0SublevelInfo not created for compaction out of L0")
    1006              :                         }
    1007            2 :                         for _, info := range c.startLevel.l0SublevelInfo {
    1008            2 :                                 err := manifest.CheckOrdering(c.comparer, info.sublevel, info.Iter())
    1009            2 :                                 if err != nil {
    1010            1 :                                         return nil, nil, nil, err
    1011            1 :                                 }
    1012              :                         }
    1013              :                 }
    1014            2 :                 if len(c.extraLevels) > 0 {
    1015            2 :                         if len(c.extraLevels) > 1 {
    1016            0 :                                 panic("n>2 multi level compaction not implemented yet")
    1017              :                         }
    1018            2 :                         interLevel := c.extraLevels[0]
    1019            2 :                         err := manifest.CheckOrdering(c.comparer, manifest.Level(interLevel.level),
    1020            2 :                                 interLevel.files.Iter())
    1021            2 :                         if err != nil {
    1022            0 :                                 return nil, nil, nil, err
    1023            0 :                         }
    1024              :                 }
    1025              :         }
    1026              : 
    1027              :         // There are three classes of keys that a compaction needs to process: point
    1028              :         // keys, range deletion tombstones and range keys. Collect all iterators for
    1029              :         // all these classes of keys from all the levels. We'll aggregate them
    1030              :         // together farther below.
    1031              :         //
    1032              :         // numInputLevels is an approximation of the number of iterator levels. Due
    1033              :         // to idiosyncrasies in iterator construction, we may (rarely) exceed this
    1034              :         // initial capacity.
    1035            2 :         numInputLevels := max(len(c.flush.flushables), len(c.inputs))
    1036            2 :         iters := make([]internalIterator, 0, numInputLevels)
    1037            2 :         rangeDelIters := make([]keyspan.FragmentIterator, 0, numInputLevels)
    1038            2 :         rangeKeyIters := make([]keyspan.FragmentIterator, 0, numInputLevels)
    1039            2 : 
    1040            2 :         // If construction of the iterator inputs fails, ensure that we close all
    1041            2 :         // the consitutent iterators.
    1042            2 :         defer func() {
    1043            2 :                 if retErr != nil {
    1044            1 :                         for _, iter := range iters {
    1045            1 :                                 if iter != nil {
    1046            1 :                                         _ = iter.Close()
    1047            1 :                                 }
    1048              :                         }
    1049            1 :                         for _, rangeDelIter := range rangeDelIters {
    1050            0 :                                 rangeDelIter.Close()
    1051            0 :                         }
    1052              :                 }
    1053              :         }()
    1054            2 :         iterOpts := IterOptions{
    1055            2 :                 Category: categoryCompaction,
    1056            2 :                 logger:   c.logger,
    1057            2 :         }
    1058            2 : 
    1059            2 :         // Populate iters, rangeDelIters and rangeKeyIters with the appropriate
    1060            2 :         // constituent iterators. This depends on whether this is a flush or a
    1061            2 :         // compaction.
    1062            2 :         if len(c.flush.flushables) != 0 {
    1063            2 :                 // If flushing, we need to build the input iterators over the memtables
    1064            2 :                 // stored in c.flush.flushables.
    1065            2 :                 for _, f := range c.flush.flushables {
    1066            2 :                         iters = append(iters, f.newFlushIter(nil))
    1067            2 :                         rangeDelIter := f.newRangeDelIter(nil)
    1068            2 :                         if rangeDelIter != nil {
    1069            2 :                                 rangeDelIters = append(rangeDelIters, rangeDelIter)
    1070            2 :                         }
    1071            2 :                         if rangeKeyIter := f.newRangeKeyIter(nil); rangeKeyIter != nil {
    1072            2 :                                 rangeKeyIters = append(rangeKeyIters, rangeKeyIter)
    1073            2 :                         }
    1074              :                 }
    1075            2 :         } else {
    1076            2 :                 addItersForLevel := func(level *compactionLevel, l manifest.Layer) error {
    1077            2 :                         // Add a *levelIter for point iterators. Because we don't call
    1078            2 :                         // initRangeDel, the levelIter will close and forget the range
    1079            2 :                         // deletion iterator when it steps on to a new file. Surfacing range
    1080            2 :                         // deletions to compactions are handled below.
    1081            2 :                         iters = append(iters, newLevelIter(ctx, iterOpts, c.comparer,
    1082            2 :                                 newIters, level.files.Iter(), l, iiopts))
    1083            2 :                         // TODO(jackson): Use keyspanimpl.LevelIter to avoid loading all the range
    1084            2 :                         // deletions into memory upfront. (See #2015, which reverted this.) There
    1085            2 :                         // will be no user keys that are split between sstables within a level in
    1086            2 :                         // Cockroach 23.1, which unblocks this optimization.
    1087            2 : 
    1088            2 :                         // Add the range deletion iterator for each file as an independent level
    1089            2 :                         // in mergingIter, as opposed to making a levelIter out of those. This
    1090            2 :                         // is safer as levelIter expects all keys coming from underlying
    1091            2 :                         // iterators to be in order. Due to compaction / tombstone writing
    1092            2 :                         // logic in finishOutput(), it is possible for range tombstones to not
    1093            2 :                         // be strictly ordered across all files in one level.
    1094            2 :                         //
    1095            2 :                         // Consider this example from the metamorphic tests (also repeated in
    1096            2 :                         // finishOutput()), consisting of three L3 files with their bounds
    1097            2 :                         // specified in square brackets next to the file name:
    1098            2 :                         //
    1099            2 :                         // ./000240.sst   [tmgc#391,MERGE-tmgc#391,MERGE]
    1100            2 :                         // tmgc#391,MERGE [786e627a]
    1101            2 :                         // tmgc-udkatvs#331,RANGEDEL
    1102            2 :                         //
    1103            2 :                         // ./000241.sst   [tmgc#384,MERGE-tmgc#384,MERGE]
    1104            2 :                         // tmgc#384,MERGE [666c7070]
    1105            2 :                         // tmgc-tvsalezade#383,RANGEDEL
    1106            2 :                         // tmgc-tvsalezade#331,RANGEDEL
    1107            2 :                         //
    1108            2 :                         // ./000242.sst   [tmgc#383,RANGEDEL-tvsalezade#72057594037927935,RANGEDEL]
    1109            2 :                         // tmgc-tvsalezade#383,RANGEDEL
    1110            2 :                         // tmgc#375,SET [72646c78766965616c72776865676e79]
    1111            2 :                         // tmgc-tvsalezade#356,RANGEDEL
    1112            2 :                         //
    1113            2 :                         // Here, the range tombstone in 000240.sst falls "after" one in
    1114            2 :                         // 000241.sst, despite 000240.sst being ordered "before" 000241.sst for
    1115            2 :                         // levelIter's purposes. While each file is still consistent before its
    1116            2 :                         // bounds, it's safer to have all rangedel iterators be visible to
    1117            2 :                         // mergingIter.
    1118            2 :                         iter := level.files.Iter()
    1119            2 :                         for f := iter.First(); f != nil; f = iter.Next() {
    1120            2 :                                 rangeDelIter, err := c.newRangeDelIter(ctx, newIters, iter.Take(), iterOpts, iiopts, l)
    1121            2 :                                 if err != nil {
    1122            1 :                                         // The error will already be annotated with the BackingFileNum, so
    1123            1 :                                         // we annotate it with the FileNum.
    1124            1 :                                         return errors.Wrapf(err, "pebble: could not open table %s", errors.Safe(f.TableNum))
    1125            1 :                                 }
    1126            2 :                                 if rangeDelIter == nil {
    1127            2 :                                         continue
    1128              :                                 }
    1129            2 :                                 rangeDelIters = append(rangeDelIters, rangeDelIter)
    1130            2 :                                 c.iterationState.keyspanIterClosers = append(c.iterationState.keyspanIterClosers, rangeDelIter)
    1131              :                         }
    1132              : 
    1133              :                         // Check if this level has any range keys.
    1134            2 :                         hasRangeKeys := false
    1135            2 :                         for f := iter.First(); f != nil; f = iter.Next() {
    1136            2 :                                 if f.HasRangeKeys {
    1137            2 :                                         hasRangeKeys = true
    1138            2 :                                         break
    1139              :                                 }
    1140              :                         }
    1141            2 :                         if hasRangeKeys {
    1142            2 :                                 newRangeKeyIterWrapper := func(ctx context.Context, file *manifest.TableMetadata, iterOptions keyspan.SpanIterOptions) (keyspan.FragmentIterator, error) {
    1143            2 :                                         iters, err := newIters(ctx, file, &iterOpts, iiopts, iterRangeKeys)
    1144            2 :                                         if err != nil {
    1145            0 :                                                 return nil, err
    1146            2 :                                         } else if iters.rangeKey == nil {
    1147            0 :                                                 return emptyKeyspanIter, nil
    1148            0 :                                         }
    1149              :                                         // Ensure that the range key iter is not closed until the compaction is
    1150              :                                         // finished. This is necessary because range key processing
    1151              :                                         // requires the range keys to be held in memory for up to the
    1152              :                                         // lifetime of the compaction.
    1153            2 :                                         noCloseIter := &noCloseIter{iters.rangeKey}
    1154            2 :                                         c.iterationState.keyspanIterClosers = append(c.iterationState.keyspanIterClosers, noCloseIter)
    1155            2 : 
    1156            2 :                                         // We do not need to truncate range keys to sstable boundaries, or
    1157            2 :                                         // only read within the file's atomic compaction units, unlike with
    1158            2 :                                         // range tombstones. This is because range keys were added after we
    1159            2 :                                         // stopped splitting user keys across sstables, so all the range keys
    1160            2 :                                         // in this sstable must wholly lie within the file's bounds.
    1161            2 :                                         return noCloseIter, err
    1162              :                                 }
    1163            2 :                                 li := keyspanimpl.NewLevelIter(ctx, keyspan.SpanIterOptions{}, cmp,
    1164            2 :                                         newRangeKeyIterWrapper, level.files.Iter(), l, manifest.KeyTypeRange)
    1165            2 :                                 rangeKeyIters = append(rangeKeyIters, li)
    1166              :                         }
    1167            2 :                         return nil
    1168              :                 }
    1169              : 
    1170            2 :                 for i := range c.inputs {
    1171            2 :                         // If the level is annotated with l0SublevelInfo, expand it into one
    1172            2 :                         // level per sublevel.
    1173            2 :                         // TODO(jackson): Perform this expansion even earlier when we pick the
    1174            2 :                         // compaction?
    1175            2 :                         if len(c.inputs[i].l0SublevelInfo) > 0 {
    1176            2 :                                 for _, info := range c.startLevel.l0SublevelInfo {
    1177            2 :                                         sublevelCompactionLevel := &compactionLevel{0, info.LevelSlice, nil}
    1178            2 :                                         if err := addItersForLevel(sublevelCompactionLevel, info.sublevel); err != nil {
    1179            1 :                                                 return nil, nil, nil, err
    1180            1 :                                         }
    1181              :                                 }
    1182            2 :                                 continue
    1183              :                         }
    1184            2 :                         if err := addItersForLevel(&c.inputs[i], manifest.Level(c.inputs[i].level)); err != nil {
    1185            1 :                                 return nil, nil, nil, err
    1186            1 :                         }
    1187              :                 }
    1188              :         }
    1189              : 
    1190              :         // If there's only one constituent point iterator, we can avoid the overhead
    1191              :         // of a *mergingIter. This is possible, for example, when performing a flush
    1192              :         // of a single memtable. Otherwise, combine all the iterators into a merging
    1193              :         // iter.
    1194            2 :         pointIter = iters[0]
    1195            2 :         if len(iters) > 1 {
    1196            2 :                 pointIter = newMergingIter(c.logger, &c.metrics.internalIterStats, cmp, nil, iters...)
    1197            2 :         }
    1198              : 
    1199              :         // In normal operation, levelIter iterates over the point operations in a
    1200              :         // level, and initializes a rangeDelIter pointer for the range deletions in
    1201              :         // each table. During compaction, we want to iterate over the merged view of
    1202              :         // point operations and range deletions. In order to do this we create one
    1203              :         // levelIter per level to iterate over the point operations, and collect up
    1204              :         // all the range deletion files.
    1205              :         //
    1206              :         // The range deletion levels are combined with a keyspanimpl.MergingIter. The
    1207              :         // resulting merged rangedel iterator is then included using an
    1208              :         // InterleavingIter.
    1209              :         // TODO(jackson): Consider using a defragmenting iterator to stitch together
    1210              :         // logical range deletions that were fragmented due to previous file
    1211              :         // boundaries.
    1212            2 :         if len(rangeDelIters) > 0 {
    1213            2 :                 mi := &keyspanimpl.MergingIter{}
    1214            2 :                 mi.Init(c.comparer, keyspan.NoopTransform, new(keyspanimpl.MergingBuffers), rangeDelIters...)
    1215            2 :                 rangeDelIter = mi
    1216            2 :         }
    1217              : 
    1218              :         // If there are range key iterators, we need to combine them using
    1219              :         // keyspanimpl.MergingIter, and then interleave them among the points.
    1220            2 :         if len(rangeKeyIters) > 0 {
    1221            2 :                 mi := &keyspanimpl.MergingIter{}
    1222            2 :                 mi.Init(c.comparer, keyspan.NoopTransform, new(keyspanimpl.MergingBuffers), rangeKeyIters...)
    1223            2 :                 // TODO(radu): why do we have a defragmenter here but not above?
    1224            2 :                 di := &keyspan.DefragmentingIter{}
    1225            2 :                 di.Init(c.comparer, mi, keyspan.DefragmentInternal, keyspan.StaticDefragmentReducer, new(keyspan.DefragmentingBuffers))
    1226            2 :                 rangeKeyIter = di
    1227            2 :         }
    1228            2 :         return pointIter, rangeDelIter, rangeKeyIter, nil
    1229              : }
    1230              : 
    1231              : func (c *tableCompaction) newRangeDelIter(
    1232              :         ctx context.Context,
    1233              :         newIters tableNewIters,
    1234              :         f manifest.LevelFile,
    1235              :         opts IterOptions,
    1236              :         iiopts internalIterOpts,
    1237              :         l manifest.Layer,
    1238            2 : ) (*noCloseIter, error) {
    1239            2 :         opts.layer = l
    1240            2 :         iterSet, err := newIters(ctx, f.TableMetadata, &opts, iiopts, iterRangeDeletions)
    1241            2 :         if err != nil {
    1242            1 :                 return nil, err
    1243            2 :         } else if iterSet.rangeDeletion == nil {
    1244            2 :                 // The file doesn't contain any range deletions.
    1245            2 :                 return nil, nil
    1246            2 :         }
    1247              :         // Ensure that rangeDelIter is not closed until the compaction is
    1248              :         // finished. This is necessary because range tombstone processing
    1249              :         // requires the range tombstones to be held in memory for up to the
    1250              :         // lifetime of the compaction.
    1251            2 :         return &noCloseIter{iterSet.rangeDeletion}, nil
    1252              : }
    1253              : 
    1254            1 : func (c *tableCompaction) String() string {
    1255            1 :         if len(c.flush.flushables) != 0 {
    1256            0 :                 return "flush\n"
    1257            0 :         }
    1258              : 
    1259            1 :         var buf bytes.Buffer
    1260            1 :         for level := c.startLevel.level; level <= c.outputLevel.level; level++ {
    1261            1 :                 i := level - c.startLevel.level
    1262            1 :                 fmt.Fprintf(&buf, "%d:", level)
    1263            1 :                 for f := range c.inputs[i].files.All() {
    1264            1 :                         fmt.Fprintf(&buf, " %s:%s-%s", f.TableNum, f.Smallest(), f.Largest())
    1265            1 :                 }
    1266            1 :                 fmt.Fprintf(&buf, "\n")
    1267              :         }
    1268            1 :         return buf.String()
    1269              : }
    1270              : 
    1271              : type manualCompaction struct {
    1272              :         // id is for internal bookkeeping.
    1273              :         id uint64
    1274              :         // Count of the retries due to concurrent compaction to overlapping levels.
    1275              :         retries     int
    1276              :         level       int
    1277              :         outputLevel int
    1278              :         done        chan error
    1279              :         start       []byte
    1280              :         end         []byte
    1281              :         split       bool
    1282              : }
    1283              : 
    1284              : type readCompaction struct {
    1285              :         level int
    1286              :         // [start, end] key ranges are used for de-duping.
    1287              :         start []byte
    1288              :         end   []byte
    1289              : 
    1290              :         // The file associated with the compaction.
    1291              :         // If the file no longer belongs in the same
    1292              :         // level, then we skip the compaction.
    1293              :         tableNum base.TableNum
    1294              : }
    1295              : 
    1296              : // Removes compaction markers from files in a compaction. The rollback parameter
    1297              : // indicates whether the compaction state should be rolled back to its original
    1298              : // state in the case of an unsuccessful compaction.
    1299              : //
    1300              : // DB.mu must be held when calling this method, however this method can drop and
    1301              : // re-acquire that mutex. All writes to the manifest for this compaction should
    1302              : // have completed by this point.
    1303            2 : func (d *DB) clearCompactingState(c *tableCompaction, rollback bool) {
    1304            2 :         c.versionEditApplied = true
    1305            2 :         for _, cl := range c.inputs {
    1306            2 :                 for f := range cl.files.All() {
    1307            2 :                         if !f.IsCompacting() {
    1308            0 :                                 d.opts.Logger.Fatalf("L%d->L%d: %s not being compacted", c.startLevel.level, c.outputLevel.level, f.TableNum)
    1309            0 :                         }
    1310            2 :                         if !rollback {
    1311            2 :                                 // On success all compactions other than move and delete-only compactions
    1312            2 :                                 // transition the file into the Compacted state. Move-compacted files
    1313            2 :                                 // become eligible for compaction again and transition back to NotCompacting.
    1314            2 :                                 // Delete-only compactions could, on rare occasion, leave files untouched
    1315            2 :                                 // (eg. if files have a loose bound), so we revert them all to NotCompacting
    1316            2 :                                 // just in case they need to be compacted again.
    1317            2 :                                 if c.kind != compactionKindMove && c.kind != compactionKindDeleteOnly {
    1318            2 :                                         f.SetCompactionState(manifest.CompactionStateCompacted)
    1319            2 :                                 } else {
    1320            2 :                                         f.SetCompactionState(manifest.CompactionStateNotCompacting)
    1321            2 :                                 }
    1322            2 :                         } else {
    1323            2 :                                 // Else, on rollback, all input files unconditionally transition back to
    1324            2 :                                 // NotCompacting.
    1325            2 :                                 f.SetCompactionState(manifest.CompactionStateNotCompacting)
    1326            2 :                         }
    1327            2 :                         f.IsIntraL0Compacting = false
    1328              :                 }
    1329              :         }
    1330            2 :         l0InProgress := inProgressL0Compactions(d.getInProgressCompactionInfoLocked(c))
    1331            2 :         func() {
    1332            2 :                 // InitCompactingFileInfo requires that no other manifest writes be
    1333            2 :                 // happening in parallel with it, i.e. we're not in the midst of installing
    1334            2 :                 // another version. Otherwise, it's possible that we've created another
    1335            2 :                 // L0Sublevels instance, but not added it to the versions list, causing
    1336            2 :                 // all the indices in TableMetadata to be inaccurate. To ensure this,
    1337            2 :                 // grab the manifest lock.
    1338            2 :                 d.mu.versions.logLock()
    1339            2 :                 // It is a bit peculiar that we are fiddling with th current version state
    1340            2 :                 // in a separate critical section from when this version was installed.
    1341            2 :                 // But this fiddling is necessary if the compaction failed. When the
    1342            2 :                 // compaction succeeded, we've already done this in UpdateVersionLocked, so
    1343            2 :                 // this seems redundant. Anyway, we clear the pickedCompactionCache since we
    1344            2 :                 // may be able to pick a better compaction (though when this compaction
    1345            2 :                 // succeeded we've also cleared the cache in UpdateVersionLocked).
    1346            2 :                 defer d.mu.versions.logUnlockAndInvalidatePickedCompactionCache()
    1347            2 :                 d.mu.versions.latest.l0Organizer.InitCompactingFileInfo(l0InProgress)
    1348            2 :         }()
    1349              : }
    1350              : 
    1351            2 : func (d *DB) calculateDiskAvailableBytes() uint64 {
    1352            2 :         space, err := d.opts.FS.GetDiskUsage(d.dirname)
    1353            2 :         if err != nil {
    1354            2 :                 if !errors.Is(err, vfs.ErrUnsupported) {
    1355            1 :                         d.opts.EventListener.BackgroundError(err)
    1356            1 :                 }
    1357              :                 // Return the last value we managed to obtain.
    1358            2 :                 return d.diskAvailBytes.Load()
    1359              :         }
    1360              : 
    1361            2 :         d.lowDiskSpaceReporter.Report(space.AvailBytes, space.TotalBytes, d.opts.EventListener)
    1362            2 :         d.diskAvailBytes.Store(space.AvailBytes)
    1363            2 :         return space.AvailBytes
    1364              : }
    1365              : 
    1366              : // maybeScheduleFlush schedules a flush if necessary.
    1367              : //
    1368              : // d.mu must be held when calling this.
    1369            2 : func (d *DB) maybeScheduleFlush() {
    1370            2 :         if d.mu.compact.flushing || d.closed.Load() != nil || d.opts.ReadOnly {
    1371            2 :                 return
    1372            2 :         }
    1373            2 :         if len(d.mu.mem.queue) <= 1 {
    1374            2 :                 return
    1375            2 :         }
    1376              : 
    1377            2 :         if !d.passedFlushThreshold() {
    1378            2 :                 return
    1379            2 :         }
    1380              : 
    1381            2 :         d.mu.compact.flushing = true
    1382            2 :         go d.flush()
    1383              : }
    1384              : 
    1385            2 : func (d *DB) passedFlushThreshold() bool {
    1386            2 :         var n int
    1387            2 :         var size uint64
    1388            2 :         for ; n < len(d.mu.mem.queue)-1; n++ {
    1389            2 :                 if !d.mu.mem.queue[n].readyForFlush() {
    1390            2 :                         break
    1391              :                 }
    1392            2 :                 if d.mu.mem.queue[n].flushForced {
    1393            2 :                         // A flush was forced. Pretend the memtable size is the configured
    1394            2 :                         // size. See minFlushSize below.
    1395            2 :                         size += d.opts.MemTableSize
    1396            2 :                 } else {
    1397            2 :                         size += d.mu.mem.queue[n].totalBytes()
    1398            2 :                 }
    1399              :         }
    1400            2 :         if n == 0 {
    1401            2 :                 // None of the immutable memtables are ready for flushing.
    1402            2 :                 return false
    1403            2 :         }
    1404              : 
    1405              :         // Only flush once the sum of the queued memtable sizes exceeds half the
    1406              :         // configured memtable size. This prevents flushing of memtables at startup
    1407              :         // while we're undergoing the ramp period on the memtable size. See
    1408              :         // DB.newMemTable().
    1409            2 :         minFlushSize := d.opts.MemTableSize / 2
    1410            2 :         return size >= minFlushSize
    1411              : }
    1412              : 
    1413            2 : func (d *DB) maybeScheduleDelayedFlush(tbl *memTable, dur time.Duration) {
    1414            2 :         var mem *flushableEntry
    1415            2 :         for _, m := range d.mu.mem.queue {
    1416            2 :                 if m.flushable == tbl {
    1417            2 :                         mem = m
    1418            2 :                         break
    1419              :                 }
    1420              :         }
    1421            2 :         if mem == nil || mem.flushForced {
    1422            2 :                 return
    1423            2 :         }
    1424            2 :         deadline := d.timeNow().Add(dur)
    1425            2 :         if !mem.delayedFlushForcedAt.IsZero() && deadline.After(mem.delayedFlushForcedAt) {
    1426            2 :                 // Already scheduled to flush sooner than within `dur`.
    1427            2 :                 return
    1428            2 :         }
    1429            2 :         mem.delayedFlushForcedAt = deadline
    1430            2 :         go func() {
    1431            2 :                 timer := time.NewTimer(dur)
    1432            2 :                 defer timer.Stop()
    1433            2 : 
    1434            2 :                 select {
    1435            2 :                 case <-d.closedCh:
    1436            2 :                         return
    1437            2 :                 case <-mem.flushed:
    1438            2 :                         return
    1439            2 :                 case <-timer.C:
    1440            2 :                         d.commit.mu.Lock()
    1441            2 :                         defer d.commit.mu.Unlock()
    1442            2 :                         d.mu.Lock()
    1443            2 :                         defer d.mu.Unlock()
    1444            2 : 
    1445            2 :                         // NB: The timer may fire concurrently with a call to Close.  If a
    1446            2 :                         // Close call beat us to acquiring d.mu, d.closed holds ErrClosed,
    1447            2 :                         // and it's too late to flush anything. Otherwise, the Close call
    1448            2 :                         // will block on locking d.mu until we've finished scheduling the
    1449            2 :                         // flush and set `d.mu.compact.flushing` to true. Close will wait
    1450            2 :                         // for the current flush to complete.
    1451            2 :                         if d.closed.Load() != nil {
    1452            1 :                                 return
    1453            1 :                         }
    1454              : 
    1455            2 :                         if d.mu.mem.mutable == tbl {
    1456            2 :                                 _ = d.makeRoomForWrite(nil)
    1457            2 :                         } else {
    1458            2 :                                 mem.flushForced = true
    1459            2 :                         }
    1460            2 :                         d.maybeScheduleFlush()
    1461              :                 }
    1462              :         }()
    1463              : }
    1464              : 
    1465            2 : func (d *DB) flush() {
    1466            2 :         pprof.Do(context.Background(), flushLabels, func(context.Context) {
    1467            2 :                 flushingWorkStart := crtime.NowMono()
    1468            2 :                 d.mu.Lock()
    1469            2 :                 defer d.mu.Unlock()
    1470            2 :                 idleDuration := flushingWorkStart.Sub(d.mu.compact.noOngoingFlushStartTime)
    1471            2 :                 var bytesFlushed uint64
    1472            2 :                 var err error
    1473            2 :                 if bytesFlushed, err = d.flush1(); err != nil {
    1474            1 :                         // TODO(peter): count consecutive flush errors and backoff.
    1475            1 :                         d.opts.EventListener.BackgroundError(err)
    1476            1 :                 }
    1477            2 :                 d.mu.compact.flushing = false
    1478            2 :                 d.mu.compact.noOngoingFlushStartTime = crtime.NowMono()
    1479            2 :                 workDuration := d.mu.compact.noOngoingFlushStartTime.Sub(flushingWorkStart)
    1480            2 :                 d.mu.compact.flushWriteThroughput.Bytes += int64(bytesFlushed)
    1481            2 :                 d.mu.compact.flushWriteThroughput.WorkDuration += workDuration
    1482            2 :                 d.mu.compact.flushWriteThroughput.IdleDuration += idleDuration
    1483            2 :                 // More flush work may have arrived while we were flushing, so schedule
    1484            2 :                 // another flush if needed.
    1485            2 :                 d.maybeScheduleFlush()
    1486            2 :                 // Let the CompactionScheduler know, so that it can react immediately to
    1487            2 :                 // an increase in DB.GetAllowedWithoutPermission.
    1488            2 :                 d.opts.Experimental.CompactionScheduler.UpdateGetAllowedWithoutPermission()
    1489            2 :                 // The flush may have produced too many files in a level, so schedule a
    1490            2 :                 // compaction if needed.
    1491            2 :                 d.maybeScheduleCompaction()
    1492            2 :                 d.mu.compact.cond.Broadcast()
    1493              :         })
    1494              : }
    1495              : 
    1496              : // runIngestFlush is used to generate a flush version edit for sstables which
    1497              : // were ingested as flushables. Both DB.mu and the manifest lock must be held
    1498              : // while runIngestFlush is called.
    1499            2 : func (d *DB) runIngestFlush(c *tableCompaction) (*manifest.VersionEdit, error) {
    1500            2 :         if len(c.flush.flushables) != 1 {
    1501            0 :                 panic("pebble: ingestedFlushable must be flushed one at a time.")
    1502              :         }
    1503              : 
    1504              :         // Finding the target level for ingestion must use the latest version
    1505              :         // after the logLock has been acquired.
    1506            2 :         version := d.mu.versions.currentVersion()
    1507            2 : 
    1508            2 :         baseLevel := d.mu.versions.picker.getBaseLevel()
    1509            2 :         ve := &manifest.VersionEdit{}
    1510            2 :         var ingestSplitFiles []ingestSplitFile
    1511            2 :         ingestFlushable := c.flush.flushables[0].flushable.(*ingestedFlushable)
    1512            2 : 
    1513            2 :         updateLevelMetricsOnExcise := func(m *manifest.TableMetadata, level int, added []manifest.NewTableEntry) {
    1514            2 :                 levelMetrics := c.metrics.perLevel[level]
    1515            2 :                 if levelMetrics == nil {
    1516            2 :                         levelMetrics = &LevelMetrics{}
    1517            2 :                         c.metrics.perLevel[level] = levelMetrics
    1518            2 :                 }
    1519            2 :                 levelMetrics.TablesCount--
    1520            2 :                 levelMetrics.TablesSize -= int64(m.Size)
    1521            2 :                 levelMetrics.EstimatedReferencesSize -= m.EstimatedReferenceSize()
    1522            2 :                 for i := range added {
    1523            2 :                         levelMetrics.TablesCount++
    1524            2 :                         levelMetrics.TablesSize += int64(added[i].Meta.Size)
    1525            2 :                         levelMetrics.EstimatedReferencesSize += added[i].Meta.EstimatedReferenceSize()
    1526            2 :                 }
    1527              :         }
    1528              : 
    1529            2 :         suggestSplit := d.opts.Experimental.IngestSplit != nil && d.opts.Experimental.IngestSplit() &&
    1530            2 :                 d.FormatMajorVersion() >= FormatVirtualSSTables
    1531            2 : 
    1532            2 :         if suggestSplit || ingestFlushable.exciseSpan.Valid() {
    1533            2 :                 // We could add deleted files to ve.
    1534            2 :                 ve.DeletedTables = make(map[manifest.DeletedTableEntry]*manifest.TableMetadata)
    1535            2 :         }
    1536              : 
    1537            2 :         ctx := context.Background()
    1538            2 :         overlapChecker := &overlapChecker{
    1539            2 :                 comparer: d.opts.Comparer,
    1540            2 :                 newIters: d.newIters,
    1541            2 :                 opts: IterOptions{
    1542            2 :                         logger:   d.opts.Logger,
    1543            2 :                         Category: categoryIngest,
    1544            2 :                 },
    1545            2 :                 v: version,
    1546            2 :         }
    1547            2 :         replacedTables := make(map[base.TableNum][]manifest.NewTableEntry)
    1548            2 :         for _, file := range ingestFlushable.files {
    1549            2 :                 var fileToSplit *manifest.TableMetadata
    1550            2 :                 var level int
    1551            2 : 
    1552            2 :                 // This file fits perfectly within the excise span, so we can slot it at L6.
    1553            2 :                 if ingestFlushable.exciseSpan.Valid() &&
    1554            2 :                         ingestFlushable.exciseSpan.Contains(d.cmp, file.Smallest()) &&
    1555            2 :                         ingestFlushable.exciseSpan.Contains(d.cmp, file.Largest()) {
    1556            2 :                         level = 6
    1557            2 :                 } else {
    1558            2 :                         // TODO(radu): this can perform I/O; we should not do this while holding DB.mu.
    1559            2 :                         lsmOverlap, err := overlapChecker.DetermineLSMOverlap(ctx, file.UserKeyBounds())
    1560            2 :                         if err != nil {
    1561            0 :                                 return nil, err
    1562            0 :                         }
    1563            2 :                         level, fileToSplit, err = ingestTargetLevel(
    1564            2 :                                 ctx, d.cmp, lsmOverlap, baseLevel, d.mu.compact.inProgress, file, suggestSplit,
    1565            2 :                         )
    1566            2 :                         if err != nil {
    1567            0 :                                 return nil, err
    1568            0 :                         }
    1569              :                 }
    1570              : 
    1571              :                 // Add the current flushableIngest file to the version.
    1572            2 :                 ve.NewTables = append(ve.NewTables, manifest.NewTableEntry{Level: level, Meta: file})
    1573            2 :                 if fileToSplit != nil {
    1574            1 :                         ingestSplitFiles = append(ingestSplitFiles, ingestSplitFile{
    1575            1 :                                 ingestFile: file,
    1576            1 :                                 splitFile:  fileToSplit,
    1577            1 :                                 level:      level,
    1578            1 :                         })
    1579            1 :                 }
    1580            2 :                 levelMetrics := c.metrics.perLevel.level(level)
    1581            2 :                 levelMetrics.TableBytesIngested += file.Size
    1582            2 :                 levelMetrics.TablesIngested++
    1583              :         }
    1584            2 :         if ingestFlushable.exciseSpan.Valid() {
    1585            2 :                 exciseBounds := ingestFlushable.exciseSpan.UserKeyBounds()
    1586            2 :                 // Iterate through all levels and find files that intersect with exciseSpan.
    1587            2 :                 for layer, ls := range version.AllLevelsAndSublevels() {
    1588            2 :                         for m := range ls.Overlaps(d.cmp, ingestFlushable.exciseSpan.UserKeyBounds()).All() {
    1589            2 :                                 leftTable, rightTable, err := d.exciseTable(context.TODO(), exciseBounds, m, layer.Level(), tightExciseBounds)
    1590            2 :                                 if err != nil {
    1591            0 :                                         return nil, err
    1592            0 :                                 }
    1593            2 :                                 newFiles := applyExciseToVersionEdit(ve, m, leftTable, rightTable, layer.Level())
    1594            2 :                                 replacedTables[m.TableNum] = newFiles
    1595            2 :                                 updateLevelMetricsOnExcise(m, layer.Level(), newFiles)
    1596              :                         }
    1597              :                 }
    1598              :         }
    1599              : 
    1600            2 :         if len(ingestSplitFiles) > 0 {
    1601            1 :                 if err := d.ingestSplit(context.TODO(), ve, updateLevelMetricsOnExcise, ingestSplitFiles, replacedTables); err != nil {
    1602            0 :                         return nil, err
    1603            0 :                 }
    1604              :         }
    1605              : 
    1606            2 :         return ve, nil
    1607              : }
    1608              : 
    1609              : // flush runs a compaction that copies the immutable memtables from memory to
    1610              : // disk.
    1611              : //
    1612              : // d.mu must be held when calling this, but the mutex may be dropped and
    1613              : // re-acquired during the course of this method.
    1614            2 : func (d *DB) flush1() (bytesFlushed uint64, err error) {
    1615            2 :         // NB: The flushable queue can contain flushables of type ingestedFlushable.
    1616            2 :         // The sstables in ingestedFlushable.files must be placed into the appropriate
    1617            2 :         // level in the lsm. Let's say the flushable queue contains a prefix of
    1618            2 :         // regular immutable memtables, then an ingestedFlushable, and then the
    1619            2 :         // mutable memtable. When the flush of the ingestedFlushable is performed,
    1620            2 :         // it needs an updated view of the lsm. That is, the prefix of immutable
    1621            2 :         // memtables must have already been flushed. Similarly, if there are two
    1622            2 :         // contiguous ingestedFlushables in the queue, then the first flushable must
    1623            2 :         // be flushed, so that the second flushable can see an updated view of the
    1624            2 :         // lsm.
    1625            2 :         //
    1626            2 :         // Given the above, we restrict flushes to either some prefix of regular
    1627            2 :         // memtables, or a single flushable of type ingestedFlushable. The DB.flush
    1628            2 :         // function will call DB.maybeScheduleFlush again, so a new flush to finish
    1629            2 :         // the remaining flush work should be scheduled right away.
    1630            2 :         //
    1631            2 :         // NB: Large batches placed in the flushable queue share the WAL with the
    1632            2 :         // previous memtable in the queue. We must ensure the property that both the
    1633            2 :         // large batch and the memtable with which it shares a WAL are flushed
    1634            2 :         // together. The property ensures that the minimum unflushed log number
    1635            2 :         // isn't incremented incorrectly. Since a flushableBatch.readyToFlush always
    1636            2 :         // returns true, and since the large batch will always be placed right after
    1637            2 :         // the memtable with which it shares a WAL, the property is naturally
    1638            2 :         // ensured. The large batch will always be placed after the memtable with
    1639            2 :         // which it shares a WAL because we ensure it in DB.commitWrite by holding
    1640            2 :         // the commitPipeline.mu and then holding DB.mu. As an extra defensive
    1641            2 :         // measure, if we try to flush the memtable without also flushing the
    1642            2 :         // flushable batch in the same flush, since the memtable and flushableBatch
    1643            2 :         // have the same logNum, the logNum invariant check below will trigger.
    1644            2 :         var n, inputs int
    1645            2 :         var inputBytes uint64
    1646            2 :         var ingest bool
    1647            2 :         for ; n < len(d.mu.mem.queue)-1; n++ {
    1648            2 :                 if f, ok := d.mu.mem.queue[n].flushable.(*ingestedFlushable); ok {
    1649            2 :                         if n == 0 {
    1650            2 :                                 // The first flushable is of type ingestedFlushable. Since these
    1651            2 :                                 // must be flushed individually, we perform a flush for just
    1652            2 :                                 // this.
    1653            2 :                                 if !f.readyForFlush() {
    1654            0 :                                         // This check is almost unnecessary, but we guard against it
    1655            0 :                                         // just in case this invariant changes in the future.
    1656            0 :                                         panic("pebble: ingestedFlushable should always be ready to flush.")
    1657              :                                 }
    1658              :                                 // By setting n = 1, we ensure that the first flushable(n == 0)
    1659              :                                 // is scheduled for a flush. The number of tables added is equal to the
    1660              :                                 // number of files in the ingest operation.
    1661            2 :                                 n = 1
    1662            2 :                                 inputs = len(f.files)
    1663            2 :                                 ingest = true
    1664            2 :                                 break
    1665            2 :                         } else {
    1666            2 :                                 // There was some prefix of flushables which weren't of type
    1667            2 :                                 // ingestedFlushable. So, perform a flush for those.
    1668            2 :                                 break
    1669              :                         }
    1670              :                 }
    1671            2 :                 if !d.mu.mem.queue[n].readyForFlush() {
    1672            1 :                         break
    1673              :                 }
    1674            2 :                 inputBytes += d.mu.mem.queue[n].inuseBytes()
    1675              :         }
    1676            2 :         if n == 0 {
    1677            0 :                 // None of the immutable memtables are ready for flushing.
    1678            0 :                 return 0, nil
    1679            0 :         }
    1680            2 :         if !ingest {
    1681            2 :                 // Flushes of memtables add the prefix of n memtables from the flushable
    1682            2 :                 // queue.
    1683            2 :                 inputs = n
    1684            2 :         }
    1685              : 
    1686              :         // Require that every memtable being flushed has a log number less than the
    1687              :         // new minimum unflushed log number.
    1688            2 :         minUnflushedLogNum := d.mu.mem.queue[n].logNum
    1689            2 :         if !d.opts.DisableWAL {
    1690            2 :                 for i := 0; i < n; i++ {
    1691            2 :                         if logNum := d.mu.mem.queue[i].logNum; logNum >= minUnflushedLogNum {
    1692            0 :                                 panic(errors.AssertionFailedf("logNum invariant violated: flushing %d items; %d:type=%T,logNum=%d; %d:type=%T,logNum=%d",
    1693            0 :                                         n,
    1694            0 :                                         i, d.mu.mem.queue[i].flushable, logNum,
    1695            0 :                                         n, d.mu.mem.queue[n].flushable, minUnflushedLogNum))
    1696              :                         }
    1697              :                 }
    1698              :         }
    1699              : 
    1700            2 :         c, err := newFlush(d.opts, d.mu.versions.currentVersion(), d.mu.versions.latest.l0Organizer,
    1701            2 :                 d.mu.versions.picker.getBaseLevel(), d.mu.mem.queue[:n], d.timeNow(), d.TableFormat(), d.determineCompactionValueSeparation)
    1702            2 :         if err != nil {
    1703            0 :                 return 0, err
    1704            0 :         }
    1705            2 :         c.AddInProgressLocked(d)
    1706            2 : 
    1707            2 :         jobID := d.newJobIDLocked()
    1708            2 :         info := FlushInfo{
    1709            2 :                 JobID:      int(jobID),
    1710            2 :                 Input:      inputs,
    1711            2 :                 InputBytes: inputBytes,
    1712            2 :                 Ingest:     ingest,
    1713            2 :         }
    1714            2 :         d.opts.EventListener.FlushBegin(info)
    1715            2 : 
    1716            2 :         startTime := d.timeNow()
    1717            2 : 
    1718            2 :         var ve *manifest.VersionEdit
    1719            2 :         var stats compact.Stats
    1720            2 :         // To determine the target level of the files in the ingestedFlushable, we
    1721            2 :         // need to acquire the logLock, and not release it for that duration. Since
    1722            2 :         // UpdateVersionLocked acquires it anyway, we create the VersionEdit for
    1723            2 :         // ingestedFlushable outside runCompaction. For all other flush cases, we
    1724            2 :         // construct the VersionEdit inside runCompaction.
    1725            2 :         var compactionErr error
    1726            2 :         if c.kind != compactionKindIngestedFlushable {
    1727            2 :                 ve, stats, compactionErr = d.runCompaction(jobID, c)
    1728            2 :         }
    1729              : 
    1730            2 :         err = d.mu.versions.UpdateVersionLocked(func() (versionUpdate, error) {
    1731            2 :                 err := compactionErr
    1732            2 :                 if c.kind == compactionKindIngestedFlushable {
    1733            2 :                         ve, err = d.runIngestFlush(c)
    1734            2 :                 }
    1735            2 :                 info.Duration = d.timeNow().Sub(startTime)
    1736            2 :                 if err != nil {
    1737            1 :                         return versionUpdate{}, err
    1738            1 :                 }
    1739              : 
    1740            2 :                 validateVersionEdit(ve, d.opts.Comparer.ValidateKey, d.opts.Comparer.FormatKey, d.opts.Logger)
    1741            2 :                 for i := range ve.NewTables {
    1742            2 :                         e := &ve.NewTables[i]
    1743            2 :                         info.Output = append(info.Output, e.Meta.TableInfo())
    1744            2 :                         // Ingested tables are not necessarily flushed to L0. Record the level of
    1745            2 :                         // each ingested file explicitly.
    1746            2 :                         if ingest {
    1747            2 :                                 info.IngestLevels = append(info.IngestLevels, e.Level)
    1748            2 :                         }
    1749              :                 }
    1750              : 
    1751              :                 // The flush succeeded or it produced an empty sstable. In either case we
    1752              :                 // want to bump the minimum unflushed log number to the log number of the
    1753              :                 // oldest unflushed memtable.
    1754            2 :                 ve.MinUnflushedLogNum = minUnflushedLogNum
    1755            2 :                 if c.kind != compactionKindIngestedFlushable {
    1756            2 :                         l0Metrics := c.metrics.perLevel.level(0)
    1757            2 :                         if d.opts.DisableWAL {
    1758            2 :                                 // If the WAL is disabled, every flushable has a zero [logSize],
    1759            2 :                                 // resulting in zero bytes in. Instead, use the number of bytes we
    1760            2 :                                 // flushed as the BytesIn. This ensures we get a reasonable w-amp
    1761            2 :                                 // calculation even when the WAL is disabled.
    1762            2 :                                 l0Metrics.TableBytesIn = l0Metrics.TableBytesFlushed + l0Metrics.BlobBytesFlushed
    1763            2 :                         } else {
    1764            2 :                                 for i := 0; i < n; i++ {
    1765            2 :                                         l0Metrics.TableBytesIn += d.mu.mem.queue[i].logSize
    1766            2 :                                 }
    1767              :                         }
    1768            2 :                 } else {
    1769            2 :                         // c.kind == compactionKindIngestedFlushable && we could have deleted files due
    1770            2 :                         // to ingest-time splits or excises.
    1771            2 :                         ingestFlushable := c.flush.flushables[0].flushable.(*ingestedFlushable)
    1772            2 :                         exciseBounds := ingestFlushable.exciseSpan.UserKeyBounds()
    1773            2 :                         for c2 := range d.mu.compact.inProgress {
    1774            2 :                                 // Check if this compaction overlaps with the excise span. Note that just
    1775            2 :                                 // checking if the inputs individually overlap with the excise span
    1776            2 :                                 // isn't sufficient; for instance, a compaction could have [a,b] and [e,f]
    1777            2 :                                 // as inputs and write it all out as [a,b,e,f] in one sstable. If we're
    1778            2 :                                 // doing a [c,d) excise at the same time as this compaction, we will have
    1779            2 :                                 // to error out the whole compaction as we can't guarantee it hasn't/won't
    1780            2 :                                 // write a file overlapping with the excise span.
    1781            2 :                                 bounds := c2.Bounds()
    1782            2 :                                 if bounds != nil && bounds.Overlaps(d.cmp, &exciseBounds) {
    1783            2 :                                         c2.Cancel()
    1784            2 :                                 }
    1785              :                         }
    1786              : 
    1787            2 :                         if len(ve.DeletedTables) > 0 {
    1788            2 :                                 // Iterate through all other compactions, and check if their inputs have
    1789            2 :                                 // been replaced due to an ingest-time split or excise. In that case,
    1790            2 :                                 // cancel the compaction.
    1791            2 :                                 for c2 := range d.mu.compact.inProgress {
    1792            2 :                                         for level, table := range c2.Tables() {
    1793            2 :                                                 if _, ok := ve.DeletedTables[manifest.DeletedTableEntry{FileNum: table.TableNum, Level: level}]; ok {
    1794            2 :                                                         c2.Cancel()
    1795            2 :                                                         break
    1796              :                                                 }
    1797              :                                         }
    1798              :                                 }
    1799              :                         }
    1800              :                 }
    1801            2 :                 return versionUpdate{
    1802            2 :                         VE:                      ve,
    1803            2 :                         JobID:                   jobID,
    1804            2 :                         Metrics:                 c.metrics.perLevel,
    1805            2 :                         InProgressCompactionsFn: func() []compactionInfo { return d.getInProgressCompactionInfoLocked(c) },
    1806              :                 }, nil
    1807              :         })
    1808              : 
    1809              :         // If err != nil, then the flush will be retried, and we will recalculate
    1810              :         // these metrics.
    1811            2 :         if err == nil {
    1812            2 :                 d.mu.snapshots.cumulativePinnedCount += stats.CumulativePinnedKeys
    1813            2 :                 d.mu.snapshots.cumulativePinnedSize += stats.CumulativePinnedSize
    1814            2 :                 d.mu.versions.metrics.Keys.MissizedTombstonesCount += stats.CountMissizedDels
    1815            2 :         }
    1816              : 
    1817            2 :         d.clearCompactingState(c, err != nil)
    1818            2 :         delete(d.mu.compact.inProgress, c)
    1819            2 :         d.mu.versions.incrementCompactions(c.kind, c.extraLevels, c.metrics.picker,
    1820            2 :                 c.metrics.bytesWritten.Load(), err)
    1821            2 : 
    1822            2 :         var flushed flushableList
    1823            2 :         if err == nil {
    1824            2 :                 flushed = d.mu.mem.queue[:n]
    1825            2 :                 d.mu.mem.queue = d.mu.mem.queue[n:]
    1826            2 :                 d.updateReadStateLocked(d.opts.DebugCheck)
    1827            2 :                 d.updateTableStatsLocked(ve.NewTables)
    1828            2 :                 if ingest {
    1829            2 :                         d.mu.versions.metrics.Flush.AsIngestCount++
    1830            2 :                         for _, l := range c.metrics.perLevel {
    1831            2 :                                 if l != nil {
    1832            2 :                                         d.mu.versions.metrics.Flush.AsIngestBytes += l.TableBytesIngested
    1833            2 :                                         d.mu.versions.metrics.Flush.AsIngestTableCount += l.TablesIngested
    1834            2 :                                 }
    1835              :                         }
    1836              :                 }
    1837            2 :                 d.maybeTransitionSnapshotsToFileOnlyLocked()
    1838              :         }
    1839              :         // Signal FlushEnd after installing the new readState. This helps for unit
    1840              :         // tests that use the callback to trigger a read using an iterator with
    1841              :         // IterOptions.OnlyReadGuaranteedDurable.
    1842            2 :         info.Err = err
    1843            2 :         if info.Err == nil && len(ve.NewTables) == 0 {
    1844            2 :                 info.Err = errEmptyTable
    1845            2 :         }
    1846            2 :         info.Done = true
    1847            2 :         info.TotalDuration = d.timeNow().Sub(startTime)
    1848            2 :         d.opts.EventListener.FlushEnd(info)
    1849            2 : 
    1850            2 :         // The order of these operations matters here for ease of testing.
    1851            2 :         // Removing the reader reference first allows tests to be guaranteed that
    1852            2 :         // the memtable reservation has been released by the time a synchronous
    1853            2 :         // flush returns. readerUnrefLocked may also produce obsolete files so the
    1854            2 :         // call to deleteObsoleteFiles must happen after it.
    1855            2 :         for i := range flushed {
    1856            2 :                 flushed[i].readerUnrefLocked(true)
    1857            2 :         }
    1858              : 
    1859            2 :         d.deleteObsoleteFiles(jobID)
    1860            2 : 
    1861            2 :         // Mark all the memtables we flushed as flushed.
    1862            2 :         for i := range flushed {
    1863            2 :                 close(flushed[i].flushed)
    1864            2 :         }
    1865              : 
    1866            2 :         return inputBytes, err
    1867              : }
    1868              : 
    1869              : // maybeTransitionSnapshotsToFileOnlyLocked transitions any "eventually
    1870              : // file-only" snapshots to be file-only if all their visible state has been
    1871              : // flushed to sstables.
    1872              : //
    1873              : // REQUIRES: d.mu.
    1874            2 : func (d *DB) maybeTransitionSnapshotsToFileOnlyLocked() {
    1875            2 :         earliestUnflushedSeqNum := d.getEarliestUnflushedSeqNumLocked()
    1876            2 :         currentVersion := d.mu.versions.currentVersion()
    1877            2 :         for s := d.mu.snapshots.root.next; s != &d.mu.snapshots.root; {
    1878            2 :                 if s.efos == nil {
    1879            2 :                         s = s.next
    1880            2 :                         continue
    1881              :                 }
    1882            2 :                 overlapsFlushable := false
    1883            2 :                 if base.Visible(earliestUnflushedSeqNum, s.efos.seqNum, base.SeqNumMax) {
    1884            2 :                         // There are some unflushed keys that are still visible to the EFOS.
    1885            2 :                         // Check if any memtables older than the EFOS contain keys within a
    1886            2 :                         // protected range of the EFOS. If no, we can transition.
    1887            2 :                         protectedRanges := make([]bounded, len(s.efos.protectedRanges))
    1888            2 :                         for i := range s.efos.protectedRanges {
    1889            2 :                                 protectedRanges[i] = s.efos.protectedRanges[i]
    1890            2 :                         }
    1891            2 :                         for i := range d.mu.mem.queue {
    1892            2 :                                 if !base.Visible(d.mu.mem.queue[i].logSeqNum, s.efos.seqNum, base.SeqNumMax) {
    1893            1 :                                         // All keys in this memtable are newer than the EFOS. Skip this
    1894            1 :                                         // memtable.
    1895            1 :                                         continue
    1896              :                                 }
    1897              :                                 // NB: computePossibleOverlaps could have false positives, such as if
    1898              :                                 // the flushable is a flushable ingest and not a memtable. In that
    1899              :                                 // case we don't open the sstables to check; we just pessimistically
    1900              :                                 // assume an overlap.
    1901            2 :                                 d.mu.mem.queue[i].computePossibleOverlaps(func(b bounded) shouldContinue {
    1902            2 :                                         overlapsFlushable = true
    1903            2 :                                         return stopIteration
    1904            2 :                                 }, protectedRanges...)
    1905            2 :                                 if overlapsFlushable {
    1906            2 :                                         break
    1907              :                                 }
    1908              :                         }
    1909              :                 }
    1910            2 :                 if overlapsFlushable {
    1911            2 :                         s = s.next
    1912            2 :                         continue
    1913              :                 }
    1914            2 :                 currentVersion.Ref()
    1915            2 : 
    1916            2 :                 // NB: s.efos.transitionToFileOnlySnapshot could close s, in which
    1917            2 :                 // case s.next would be nil. Save it before calling it.
    1918            2 :                 next := s.next
    1919            2 :                 _ = s.efos.transitionToFileOnlySnapshot(currentVersion)
    1920            2 :                 s = next
    1921              :         }
    1922              : }
    1923              : 
    1924              : // maybeScheduleCompactionAsync should be used when
    1925              : // we want to possibly schedule a compaction, but don't
    1926              : // want to eat the cost of running maybeScheduleCompaction.
    1927              : // This method should be launched in a separate goroutine.
    1928              : // d.mu must not be held when this is called.
    1929            0 : func (d *DB) maybeScheduleCompactionAsync() {
    1930            0 :         defer d.compactionSchedulers.Done()
    1931            0 : 
    1932            0 :         d.mu.Lock()
    1933            0 :         d.maybeScheduleCompaction()
    1934            0 :         d.mu.Unlock()
    1935            0 : }
    1936              : 
    1937              : // maybeScheduleCompaction schedules a compaction if necessary.
    1938              : //
    1939              : // WARNING: maybeScheduleCompaction and Schedule must be the only ways that
    1940              : // any compactions are run. These ensure that the pickedCompactionCache is
    1941              : // used and not stale (by ensuring invalidation is done).
    1942              : //
    1943              : // Even compactions that are not scheduled by the CompactionScheduler must be
    1944              : // run using maybeScheduleCompaction, since starting those compactions needs
    1945              : // to invalidate the pickedCompactionCache.
    1946              : //
    1947              : // Requires d.mu to be held.
    1948            2 : func (d *DB) maybeScheduleCompaction() {
    1949            2 :         d.mu.versions.logLock()
    1950            2 :         defer d.mu.versions.logUnlock()
    1951            2 :         env := d.makeCompactionEnvLocked()
    1952            2 :         if env == nil {
    1953            2 :                 return
    1954            2 :         }
    1955              :         // env.inProgressCompactions will become stale once we pick a compaction, so
    1956              :         // it needs to be kept fresh. Also, the pickedCompaction in the
    1957              :         // pickedCompactionCache is not valid if we pick a compaction before using
    1958              :         // it, since those earlier compactions can mark the same file as compacting.
    1959              : 
    1960              :         // Delete-only compactions are expected to be cheap and reduce future
    1961              :         // compaction work, so schedule them directly instead of using the
    1962              :         // CompactionScheduler.
    1963            2 :         if d.tryScheduleDeleteOnlyCompaction() {
    1964            2 :                 env.inProgressCompactions = d.getInProgressCompactionInfoLocked(nil)
    1965            2 :                 d.mu.versions.pickedCompactionCache.invalidate()
    1966            2 :         }
    1967              :         // Download compactions have their own concurrency and do not currently
    1968              :         // interact with CompactionScheduler.
    1969              :         //
    1970              :         // TODO(sumeer): integrate with CompactionScheduler, since these consume
    1971              :         // disk write bandwidth.
    1972            2 :         if d.tryScheduleDownloadCompactions(*env, d.opts.MaxConcurrentDownloads()) {
    1973            2 :                 env.inProgressCompactions = d.getInProgressCompactionInfoLocked(nil)
    1974            2 :                 d.mu.versions.pickedCompactionCache.invalidate()
    1975            2 :         }
    1976              :         // The remaining compactions are scheduled by the CompactionScheduler.
    1977            2 :         if d.mu.versions.pickedCompactionCache.isWaiting() {
    1978            2 :                 // CompactionScheduler already knows that the DB is waiting to run a
    1979            2 :                 // compaction.
    1980            2 :                 return
    1981            2 :         }
    1982              :         // INVARIANT: !pickedCompactionCache.isWaiting. The following loop will
    1983              :         // either exit after successfully starting all the compactions it can pick,
    1984              :         // or will exit with one pickedCompaction in the cache, and isWaiting=true.
    1985            2 :         for {
    1986            2 :                 // Do not have a pickedCompaction in the cache.
    1987            2 :                 pc := d.pickAnyCompaction(*env)
    1988            2 :                 if pc == nil {
    1989            2 :                         return
    1990            2 :                 }
    1991            2 :                 success, grantHandle := d.opts.Experimental.CompactionScheduler.TrySchedule()
    1992            2 :                 if !success {
    1993            2 :                         // Can't run now, but remember this pickedCompaction in the cache.
    1994            2 :                         d.mu.versions.pickedCompactionCache.add(pc)
    1995            2 :                         return
    1996            2 :                 }
    1997            2 :                 d.runPickedCompaction(pc, grantHandle)
    1998            2 :                 env.inProgressCompactions = d.getInProgressCompactionInfoLocked(nil)
    1999              :         }
    2000              : }
    2001              : 
    2002              : // makeCompactionEnv attempts to create a compactionEnv necessary during
    2003              : // compaction picking. If the DB is closed or marked as read-only,
    2004              : // makeCompactionEnv returns nil to indicate that compactions may not be
    2005              : // performed. Else, a new compactionEnv is constructed using the current DB
    2006              : // state.
    2007              : //
    2008              : // Compaction picking needs a coherent view of a Version. For example, we need
    2009              : // to exclude concurrent ingestions from making a decision on which level to
    2010              : // ingest into that conflicts with our compaction decision.
    2011              : //
    2012              : // A pickedCompaction constructed using a compactionEnv must only be used if
    2013              : // the latest Version has not changed.
    2014              : //
    2015              : // REQUIRES: d.mu and d.mu.versions.logLock are held.
    2016            2 : func (d *DB) makeCompactionEnvLocked() *compactionEnv {
    2017            2 :         if d.closed.Load() != nil || d.opts.ReadOnly {
    2018            2 :                 return nil
    2019            2 :         }
    2020            2 :         env := &compactionEnv{
    2021            2 :                 diskAvailBytes:          d.diskAvailBytes.Load(),
    2022            2 :                 earliestSnapshotSeqNum:  d.mu.snapshots.earliest(),
    2023            2 :                 earliestUnflushedSeqNum: d.getEarliestUnflushedSeqNumLocked(),
    2024            2 :                 inProgressCompactions:   d.getInProgressCompactionInfoLocked(nil),
    2025            2 :                 readCompactionEnv: readCompactionEnv{
    2026            2 :                         readCompactions:          &d.mu.compact.readCompactions,
    2027            2 :                         flushing:                 d.mu.compact.flushing || d.passedFlushThreshold(),
    2028            2 :                         rescheduleReadCompaction: &d.mu.compact.rescheduleReadCompaction,
    2029            2 :                 },
    2030            2 :         }
    2031            2 :         if !d.problemSpans.IsEmpty() {
    2032            1 :                 env.problemSpans = &d.problemSpans
    2033            1 :         }
    2034            2 :         return env
    2035              : }
    2036              : 
    2037              : // pickAnyCompaction tries to pick a manual or automatic compaction.
    2038            2 : func (d *DB) pickAnyCompaction(env compactionEnv) (pc pickedCompaction) {
    2039            2 :         // Pick a score-based compaction first, since a misshapen LSM is bad.
    2040            2 :         if !d.opts.DisableAutomaticCompactions {
    2041            2 :                 if pc = d.mu.versions.picker.pickAutoScore(env); pc != nil {
    2042            2 :                         return pc
    2043            2 :                 }
    2044              :         }
    2045              :         // Pick a manual compaction, if any.
    2046            2 :         if pc = d.pickManualCompaction(env); pc != nil {
    2047            2 :                 return pc
    2048            2 :         }
    2049            2 :         if !d.opts.DisableAutomaticCompactions {
    2050            2 :                 return d.mu.versions.picker.pickAutoNonScore(env)
    2051            2 :         }
    2052            2 :         return nil
    2053              : }
    2054              : 
    2055              : // runPickedCompaction kicks off the provided pickedCompaction. In case the
    2056              : // pickedCompaction is a manual compaction, the corresponding manualCompaction
    2057              : // is removed from d.mu.compact.manual.
    2058              : //
    2059              : // REQUIRES: d.mu and d.mu.versions.logLock is held.
    2060            2 : func (d *DB) runPickedCompaction(pc pickedCompaction, grantHandle CompactionGrantHandle) {
    2061            2 :         var doneChannel chan error
    2062            2 :         if pc.ManualID() > 0 {
    2063            2 :                 for i := range d.mu.compact.manual {
    2064            2 :                         if d.mu.compact.manual[i].id == pc.ManualID() {
    2065            2 :                                 doneChannel = d.mu.compact.manual[i].done
    2066            2 :                                 d.mu.compact.manual = slices.Delete(d.mu.compact.manual, i, i+1)
    2067            2 :                                 d.mu.compact.manualLen.Store(int32(len(d.mu.compact.manual)))
    2068            2 :                                 break
    2069              :                         }
    2070              :                 }
    2071            2 :                 if doneChannel == nil {
    2072            0 :                         panic(errors.AssertionFailedf("did not find manual compaction with id %d", pc.ManualID()))
    2073              :                 }
    2074              :         }
    2075              : 
    2076            2 :         d.mu.compact.compactingCount++
    2077            2 :         c := pc.ConstructCompaction(d, grantHandle)
    2078            2 :         c.AddInProgressLocked(d)
    2079            2 :         go func() {
    2080            2 :                 d.compact(c, doneChannel)
    2081            2 :         }()
    2082              : }
    2083              : 
    2084              : // Schedule implements DBForCompaction (it is called by the
    2085              : // CompactionScheduler).
    2086            2 : func (d *DB) Schedule(grantHandle CompactionGrantHandle) bool {
    2087            2 :         d.mu.Lock()
    2088            2 :         defer d.mu.Unlock()
    2089            2 :         d.mu.versions.logLock()
    2090            2 :         defer d.mu.versions.logUnlock()
    2091            2 :         isWaiting := d.mu.versions.pickedCompactionCache.isWaiting()
    2092            2 :         if !isWaiting {
    2093            0 :                 return false
    2094            0 :         }
    2095            2 :         pc := d.mu.versions.pickedCompactionCache.getForRunning()
    2096            2 :         if pc == nil {
    2097            2 :                 env := d.makeCompactionEnvLocked()
    2098            2 :                 if env != nil {
    2099            2 :                         pc = d.pickAnyCompaction(*env)
    2100            2 :                 }
    2101            2 :                 if pc == nil {
    2102            0 :                         d.mu.versions.pickedCompactionCache.setNotWaiting()
    2103            0 :                         return false
    2104            0 :                 }
    2105              :         }
    2106              :         // INVARIANT: pc != nil and is not in the cache. isWaiting is true, since
    2107              :         // there may be more compactions to run.
    2108            2 :         d.runPickedCompaction(pc, grantHandle)
    2109            2 :         return true
    2110              : }
    2111              : 
    2112              : // GetWaitingCompaction implements DBForCompaction (it is called by the
    2113              : // CompactionScheduler).
    2114            2 : func (d *DB) GetWaitingCompaction() (bool, WaitingCompaction) {
    2115            2 :         d.mu.Lock()
    2116            2 :         defer d.mu.Unlock()
    2117            2 :         d.mu.versions.logLock()
    2118            2 :         defer d.mu.versions.logUnlock()
    2119            2 :         isWaiting := d.mu.versions.pickedCompactionCache.isWaiting()
    2120            2 :         if !isWaiting {
    2121            2 :                 return false, WaitingCompaction{}
    2122            2 :         }
    2123            2 :         pc := d.mu.versions.pickedCompactionCache.peek()
    2124            2 :         if pc == nil {
    2125            2 :                 // Need to pick a compaction.
    2126            2 :                 env := d.makeCompactionEnvLocked()
    2127            2 :                 if env != nil {
    2128            2 :                         pc = d.pickAnyCompaction(*env)
    2129            2 :                 }
    2130            2 :                 if pc == nil {
    2131            2 :                         // Call setNotWaiting so that next call to GetWaitingCompaction can
    2132            2 :                         // return early.
    2133            2 :                         d.mu.versions.pickedCompactionCache.setNotWaiting()
    2134            2 :                         return false, WaitingCompaction{}
    2135            2 :                 } else {
    2136            2 :                         d.mu.versions.pickedCompactionCache.add(pc)
    2137            2 :                 }
    2138              :         }
    2139              :         // INVARIANT: pc != nil and is in the cache.
    2140            2 :         return true, pc.WaitingCompaction()
    2141              : }
    2142              : 
    2143              : // GetAllowedWithoutPermission implements DBForCompaction (it is called by the
    2144              : // CompactionScheduler).
    2145            2 : func (d *DB) GetAllowedWithoutPermission() int {
    2146            2 :         allowedBasedOnBacklog := int(d.mu.versions.curCompactionConcurrency.Load())
    2147            2 :         allowedBasedOnManual := 0
    2148            2 :         manualBacklog := int(d.mu.compact.manualLen.Load())
    2149            2 :         if manualBacklog > 0 {
    2150            2 :                 _, maxAllowed := d.opts.CompactionConcurrencyRange()
    2151            2 :                 allowedBasedOnManual = min(maxAllowed, manualBacklog+allowedBasedOnBacklog)
    2152            2 :         }
    2153            2 :         return max(allowedBasedOnBacklog, allowedBasedOnManual)
    2154              : }
    2155              : 
    2156              : // tryScheduleDownloadCompactions tries to start download compactions.
    2157              : //
    2158              : // Requires d.mu to be held. Updates d.mu.compact.downloads.
    2159              : //
    2160              : // Returns true iff at least one compaction was started.
    2161            2 : func (d *DB) tryScheduleDownloadCompactions(env compactionEnv, maxConcurrentDownloads int) bool {
    2162            2 :         started := false
    2163            2 :         vers := d.mu.versions.currentVersion()
    2164            2 :         for i := 0; i < len(d.mu.compact.downloads); {
    2165            2 :                 if d.mu.compact.downloadingCount >= maxConcurrentDownloads {
    2166            2 :                         break
    2167              :                 }
    2168            2 :                 download := d.mu.compact.downloads[i]
    2169            2 :                 switch d.tryLaunchDownloadCompaction(download, vers, d.mu.versions.latest.l0Organizer, env, maxConcurrentDownloads) {
    2170            2 :                 case launchedCompaction:
    2171            2 :                         started = true
    2172            2 :                         continue
    2173            1 :                 case didNotLaunchCompaction:
    2174            1 :                         // See if we can launch a compaction for another download task.
    2175            1 :                         i++
    2176            2 :                 case downloadTaskCompleted:
    2177            2 :                         // Task is completed and must be removed.
    2178            2 :                         d.mu.compact.downloads = slices.Delete(d.mu.compact.downloads, i, i+1)
    2179              :                 }
    2180              :         }
    2181            2 :         return started
    2182              : }
    2183              : 
    2184            2 : func (d *DB) pickManualCompaction(env compactionEnv) (pc pickedCompaction) {
    2185            2 :         v := d.mu.versions.currentVersion()
    2186            2 :         for len(d.mu.compact.manual) > 0 {
    2187            2 :                 manual := d.mu.compact.manual[0]
    2188            2 :                 pc, retryLater := newPickedManualCompaction(v, d.mu.versions.latest.l0Organizer,
    2189            2 :                         d.opts, env, d.mu.versions.picker.getBaseLevel(), manual)
    2190            2 :                 if pc != nil {
    2191            2 :                         return pc
    2192            2 :                 }
    2193            2 :                 if retryLater {
    2194            2 :                         // We are not able to run this manual compaction at this time.
    2195            2 :                         // Inability to run the head blocks later manual compactions.
    2196            2 :                         manual.retries++
    2197            2 :                         return nil
    2198            2 :                 }
    2199              :                 // Manual compaction is a no-op. Signal that it's complete.
    2200            2 :                 manual.done <- nil
    2201            2 :                 d.mu.compact.manual = d.mu.compact.manual[1:]
    2202            2 :                 d.mu.compact.manualLen.Store(int32(len(d.mu.compact.manual)))
    2203              :         }
    2204            2 :         return nil
    2205              : }
    2206              : 
    2207              : // tryScheduleDeleteOnlyCompaction tries to kick off a delete-only compaction
    2208              : // for all files that can be deleted as suggested by deletionHints.
    2209              : //
    2210              : // Requires d.mu to be held. Updates d.mu.compact.deletionHints.
    2211              : //
    2212              : // Returns true iff a compaction was started.
    2213            2 : func (d *DB) tryScheduleDeleteOnlyCompaction() bool {
    2214            2 :         if d.opts.private.disableDeleteOnlyCompactions || d.opts.DisableAutomaticCompactions ||
    2215            2 :                 len(d.mu.compact.deletionHints) == 0 {
    2216            2 :                 return false
    2217            2 :         }
    2218            2 :         if _, maxConcurrency := d.opts.CompactionConcurrencyRange(); d.mu.compact.compactingCount >= maxConcurrency {
    2219            2 :                 return false
    2220            2 :         }
    2221            2 :         v := d.mu.versions.currentVersion()
    2222            2 :         snapshots := d.mu.snapshots.toSlice()
    2223            2 :         // We need to save the value of exciseEnabled in the compaction itself, as
    2224            2 :         // it can change dynamically between now and when the compaction runs.
    2225            2 :         exciseEnabled := d.FormatMajorVersion() >= FormatVirtualSSTables &&
    2226            2 :                 d.opts.Experimental.EnableDeleteOnlyCompactionExcises != nil && d.opts.Experimental.EnableDeleteOnlyCompactionExcises()
    2227            2 :         inputs, resolvedHints, unresolvedHints := checkDeleteCompactionHints(d.cmp, v, d.mu.compact.deletionHints, snapshots, exciseEnabled)
    2228            2 :         d.mu.compact.deletionHints = unresolvedHints
    2229            2 : 
    2230            2 :         if len(inputs) > 0 {
    2231            2 :                 c := newDeleteOnlyCompaction(d.opts, v, inputs, d.timeNow(), resolvedHints, exciseEnabled)
    2232            2 :                 d.mu.compact.compactingCount++
    2233            2 :                 c.AddInProgressLocked(d)
    2234            2 :                 go d.compact(c, nil)
    2235            2 :                 return true
    2236            2 :         }
    2237            2 :         return false
    2238              : }
    2239              : 
    2240              : // deleteCompactionHintType indicates whether the deleteCompactionHint was
    2241              : // generated from a span containing a range del (point key only), a range key
    2242              : // delete (range key only), or both a point and range key.
    2243              : type deleteCompactionHintType uint8
    2244              : 
    2245              : const (
    2246              :         // NOTE: While these are primarily used as enumeration types, they are also
    2247              :         // used for some bitwise operations. Care should be taken when updating.
    2248              :         deleteCompactionHintTypeUnknown deleteCompactionHintType = iota
    2249              :         deleteCompactionHintTypePointKeyOnly
    2250              :         deleteCompactionHintTypeRangeKeyOnly
    2251              :         deleteCompactionHintTypePointAndRangeKey
    2252              : )
    2253              : 
    2254              : // String implements fmt.Stringer.
    2255            1 : func (h deleteCompactionHintType) String() string {
    2256            1 :         switch h {
    2257            0 :         case deleteCompactionHintTypeUnknown:
    2258            0 :                 return "unknown"
    2259            1 :         case deleteCompactionHintTypePointKeyOnly:
    2260            1 :                 return "point-key-only"
    2261            1 :         case deleteCompactionHintTypeRangeKeyOnly:
    2262            1 :                 return "range-key-only"
    2263            1 :         case deleteCompactionHintTypePointAndRangeKey:
    2264            1 :                 return "point-and-range-key"
    2265            0 :         default:
    2266            0 :                 panic(fmt.Sprintf("unknown hint type: %d", h))
    2267              :         }
    2268              : }
    2269              : 
    2270              : // compactionHintFromKeys returns a deleteCompactionHintType given a slice of
    2271              : // keyspan.Keys.
    2272            2 : func compactionHintFromKeys(keys []keyspan.Key) deleteCompactionHintType {
    2273            2 :         var hintType deleteCompactionHintType
    2274            2 :         for _, k := range keys {
    2275            2 :                 switch k.Kind() {
    2276            2 :                 case base.InternalKeyKindRangeDelete:
    2277            2 :                         hintType |= deleteCompactionHintTypePointKeyOnly
    2278            2 :                 case base.InternalKeyKindRangeKeyDelete:
    2279            2 :                         hintType |= deleteCompactionHintTypeRangeKeyOnly
    2280            0 :                 default:
    2281            0 :                         panic(fmt.Sprintf("unsupported key kind: %s", k.Kind()))
    2282              :                 }
    2283              :         }
    2284            2 :         return hintType
    2285              : }
    2286              : 
    2287              : // A deleteCompactionHint records a user key and sequence number span that has been
    2288              : // deleted by a range tombstone. A hint is recorded if at least one sstable
    2289              : // falls completely within both the user key and sequence number spans.
    2290              : // Once the tombstones and the observed completely-contained sstables fall
    2291              : // into the same snapshot stripe, a delete-only compaction may delete any
    2292              : // sstables within the range.
    2293              : type deleteCompactionHint struct {
    2294              :         // The type of key span that generated this hint (point key, range key, or
    2295              :         // both).
    2296              :         hintType deleteCompactionHintType
    2297              :         // start and end are user keys specifying a key range [start, end) of
    2298              :         // deleted keys.
    2299              :         start []byte
    2300              :         end   []byte
    2301              :         // The level of the file containing the range tombstone(s) when the hint
    2302              :         // was created. Only lower levels need to be searched for files that may
    2303              :         // be deleted.
    2304              :         tombstoneLevel int
    2305              :         // The file containing the range tombstone(s) that created the hint.
    2306              :         tombstoneFile *manifest.TableMetadata
    2307              :         // The smallest and largest sequence numbers of the abutting tombstones
    2308              :         // merged to form this hint. All of a tables' keys must be less than the
    2309              :         // tombstone smallest sequence number to be deleted. All of a tables'
    2310              :         // sequence numbers must fall into the same snapshot stripe as the
    2311              :         // tombstone largest sequence number to be deleted.
    2312              :         tombstoneLargestSeqNum  base.SeqNum
    2313              :         tombstoneSmallestSeqNum base.SeqNum
    2314              :         // The smallest sequence number of a sstable that was found to be covered
    2315              :         // by this hint. The hint cannot be resolved until this sequence number is
    2316              :         // in the same snapshot stripe as the largest tombstone sequence number.
    2317              :         // This is set when a hint is created, so the LSM may look different and
    2318              :         // notably no longer contain the sstable that contained the key at this
    2319              :         // sequence number.
    2320              :         fileSmallestSeqNum base.SeqNum
    2321              : }
    2322              : 
    2323              : type deletionHintOverlap int8
    2324              : 
    2325              : const (
    2326              :         // hintDoesNotApply indicates that the hint does not apply to the file.
    2327              :         hintDoesNotApply deletionHintOverlap = iota
    2328              :         // hintExcisesFile indicates that the hint excises a portion of the file,
    2329              :         // and the format major version of the DB supports excises.
    2330              :         hintExcisesFile
    2331              :         // hintDeletesFile indicates that the hint deletes the entirety of the file.
    2332              :         hintDeletesFile
    2333              : )
    2334              : 
    2335            1 : func (h deleteCompactionHint) String() string {
    2336            1 :         return fmt.Sprintf(
    2337            1 :                 "L%d.%s %s-%s seqnums(tombstone=%d-%d, file-smallest=%d, type=%s)",
    2338            1 :                 h.tombstoneLevel, h.tombstoneFile.TableNum, h.start, h.end,
    2339            1 :                 h.tombstoneSmallestSeqNum, h.tombstoneLargestSeqNum, h.fileSmallestSeqNum,
    2340            1 :                 h.hintType,
    2341            1 :         )
    2342            1 : }
    2343              : 
    2344              : func (h *deleteCompactionHint) canDeleteOrExcise(
    2345              :         cmp Compare, m *manifest.TableMetadata, snapshots compact.Snapshots, exciseEnabled bool,
    2346            2 : ) deletionHintOverlap {
    2347            2 :         // The file can only be deleted if all of its keys are older than the
    2348            2 :         // earliest tombstone aggregated into the hint. Note that we use
    2349            2 :         // m.LargestSeqNumAbsolute, not m.LargestSeqNum. Consider a compaction that
    2350            2 :         // zeroes sequence numbers. A compaction may zero the sequence number of a
    2351            2 :         // key with a sequence number > h.tombstoneSmallestSeqNum and set it to
    2352            2 :         // zero. If we looked at m.LargestSeqNum, the resulting output file would
    2353            2 :         // appear to not contain any keys more recent than the oldest tombstone. To
    2354            2 :         // avoid this error, the largest pre-zeroing sequence number is maintained
    2355            2 :         // in LargestSeqNumAbsolute and used here to make the determination whether
    2356            2 :         // the file's keys are older than all of the hint's tombstones.
    2357            2 :         if m.LargestSeqNumAbsolute >= h.tombstoneSmallestSeqNum || m.SmallestSeqNum < h.fileSmallestSeqNum {
    2358            2 :                 return hintDoesNotApply
    2359            2 :         }
    2360              : 
    2361              :         // The file's oldest key must  be in the same snapshot stripe as the
    2362              :         // newest tombstone. NB: We already checked the hint's sequence numbers,
    2363              :         // but this file's oldest sequence number might be lower than the hint's
    2364              :         // smallest sequence number despite the file falling within the key range
    2365              :         // if this file was constructed after the hint by a compaction.
    2366            2 :         if snapshots.Index(h.tombstoneLargestSeqNum) != snapshots.Index(m.SmallestSeqNum) {
    2367            0 :                 return hintDoesNotApply
    2368            0 :         }
    2369              : 
    2370            2 :         switch h.hintType {
    2371            2 :         case deleteCompactionHintTypePointKeyOnly:
    2372            2 :                 // A hint generated by a range del span cannot delete tables that contain
    2373            2 :                 // range keys.
    2374            2 :                 if m.HasRangeKeys {
    2375            1 :                         return hintDoesNotApply
    2376            1 :                 }
    2377            2 :         case deleteCompactionHintTypeRangeKeyOnly:
    2378            2 :                 // A hint generated by a range key del span cannot delete tables that
    2379            2 :                 // contain point keys.
    2380            2 :                 if m.HasPointKeys {
    2381            2 :                         return hintDoesNotApply
    2382            2 :                 }
    2383            2 :         case deleteCompactionHintTypePointAndRangeKey:
    2384              :                 // A hint from a span that contains both range dels *and* range keys can
    2385              :                 // only be deleted if both bounds fall within the hint. The next check takes
    2386              :                 // care of this.
    2387            0 :         default:
    2388            0 :                 panic(fmt.Sprintf("pebble: unknown delete compaction hint type: %d", h.hintType))
    2389              :         }
    2390            2 :         if cmp(h.start, m.Smallest().UserKey) <= 0 &&
    2391            2 :                 base.UserKeyExclusive(h.end).CompareUpperBounds(cmp, m.UserKeyBounds().End) >= 0 {
    2392            2 :                 return hintDeletesFile
    2393            2 :         }
    2394            2 :         if !exciseEnabled {
    2395            2 :                 // The file's keys must be completely contained within the hint range; excises
    2396            2 :                 // aren't allowed.
    2397            2 :                 return hintDoesNotApply
    2398            2 :         }
    2399              :         // Check for any overlap. In cases of partial overlap, we can excise the part of the file
    2400              :         // that overlaps with the deletion hint.
    2401            2 :         if cmp(h.end, m.Smallest().UserKey) > 0 &&
    2402            2 :                 (m.UserKeyBounds().End.CompareUpperBounds(cmp, base.UserKeyInclusive(h.start)) >= 0) {
    2403            2 :                 return hintExcisesFile
    2404            2 :         }
    2405            2 :         return hintDoesNotApply
    2406              : }
    2407              : 
    2408              : // checkDeleteCompactionHints checks the passed-in deleteCompactionHints for those that
    2409              : // can be resolved and those that cannot. A hint is considered resolved when its largest
    2410              : // tombstone sequence number and the smallest sequence number of covered files fall in
    2411              : // the same snapshot stripe. No more than maxHintsPerDeleteOnlyCompaction will be resolved
    2412              : // per method call. Resolved and unresolved hints are returned in separate return values.
    2413              : // The files that the resolved hints apply to, are returned as compactionLevels.
    2414              : func checkDeleteCompactionHints(
    2415              :         cmp Compare,
    2416              :         v *manifest.Version,
    2417              :         hints []deleteCompactionHint,
    2418              :         snapshots compact.Snapshots,
    2419              :         exciseEnabled bool,
    2420            2 : ) (levels []compactionLevel, resolved, unresolved []deleteCompactionHint) {
    2421            2 :         var files map[*manifest.TableMetadata]bool
    2422            2 :         var byLevel [numLevels][]*manifest.TableMetadata
    2423            2 : 
    2424            2 :         // Delete-only compactions can be quadratic (O(mn)) in terms of runtime
    2425            2 :         // where m = number of files in the delete-only compaction and n = number
    2426            2 :         // of resolved hints. To prevent these from growing unbounded, we cap
    2427            2 :         // the number of hints we resolve for one delete-only compaction. This
    2428            2 :         // cap only applies if exciseEnabled == true.
    2429            2 :         const maxHintsPerDeleteOnlyCompaction = 10
    2430            2 : 
    2431            2 :         unresolvedHints := hints[:0]
    2432            2 :         // Lazily populate resolvedHints, similar to files above.
    2433            2 :         resolvedHints := make([]deleteCompactionHint, 0)
    2434            2 :         for _, h := range hints {
    2435            2 :                 // Check each compaction hint to see if it's resolvable. Resolvable
    2436            2 :                 // hints are removed and trigger a delete-only compaction if any files
    2437            2 :                 // in the current LSM still meet their criteria. Unresolvable hints
    2438            2 :                 // are saved and don't trigger a delete-only compaction.
    2439            2 :                 //
    2440            2 :                 // When a compaction hint is created, the sequence numbers of the
    2441            2 :                 // range tombstones and the covered file with the oldest key are
    2442            2 :                 // recorded. The largest tombstone sequence number and the smallest
    2443            2 :                 // file sequence number must be in the same snapshot stripe for the
    2444            2 :                 // hint to be resolved. The below graphic models a compaction hint
    2445            2 :                 // covering the keyspace [b, r). The hint completely contains two
    2446            2 :                 // files, 000002 and 000003. The file 000003 contains the lowest
    2447            2 :                 // covered sequence number at #90. The tombstone b.RANGEDEL.230:h has
    2448            2 :                 // the highest tombstone sequence number incorporated into the hint.
    2449            2 :                 // The hint may be resolved only once the snapshots at #100, #180 and
    2450            2 :                 // #210 are all closed. File 000001 is not included within the hint
    2451            2 :                 // because it extends beyond the range tombstones in user key space.
    2452            2 :                 //
    2453            2 :                 // 250
    2454            2 :                 //
    2455            2 :                 //       |-b...230:h-|
    2456            2 :                 // _____________________________________________________ snapshot #210
    2457            2 :                 // 200               |--h.RANGEDEL.200:r--|
    2458            2 :                 //
    2459            2 :                 // _____________________________________________________ snapshot #180
    2460            2 :                 //
    2461            2 :                 // 150                     +--------+
    2462            2 :                 //           +---------+   | 000003 |
    2463            2 :                 //           | 000002  |   |        |
    2464            2 :                 //           +_________+   |        |
    2465            2 :                 // 100_____________________|________|___________________ snapshot #100
    2466            2 :                 //                         +--------+
    2467            2 :                 // _____________________________________________________ snapshot #70
    2468            2 :                 //                             +---------------+
    2469            2 :                 //  50                         | 000001        |
    2470            2 :                 //                             |               |
    2471            2 :                 //                             +---------------+
    2472            2 :                 // ______________________________________________________________
    2473            2 :                 //     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
    2474            2 : 
    2475            2 :                 if snapshots.Index(h.tombstoneLargestSeqNum) != snapshots.Index(h.fileSmallestSeqNum) ||
    2476            2 :                         (len(resolvedHints) >= maxHintsPerDeleteOnlyCompaction && exciseEnabled) {
    2477            2 :                         // Cannot resolve yet.
    2478            2 :                         unresolvedHints = append(unresolvedHints, h)
    2479            2 :                         continue
    2480              :                 }
    2481              : 
    2482              :                 // The hint h will be resolved and dropped, if it either affects no files at all
    2483              :                 // or if the number of files it creates (eg. through excision) is less than or
    2484              :                 // equal to the number of files it deletes. First, determine how many files are
    2485              :                 // affected by this hint.
    2486            2 :                 filesDeletedByCurrentHint := 0
    2487            2 :                 var filesDeletedByLevel [7][]*manifest.TableMetadata
    2488            2 :                 for l := h.tombstoneLevel + 1; l < numLevels; l++ {
    2489            2 :                         for m := range v.Overlaps(l, base.UserKeyBoundsEndExclusive(h.start, h.end)).All() {
    2490            2 :                                 doesHintApply := h.canDeleteOrExcise(cmp, m, snapshots, exciseEnabled)
    2491            2 :                                 if m.IsCompacting() || doesHintApply == hintDoesNotApply || files[m] {
    2492            2 :                                         continue
    2493              :                                 }
    2494            2 :                                 switch doesHintApply {
    2495            2 :                                 case hintDeletesFile:
    2496            2 :                                         filesDeletedByCurrentHint++
    2497            2 :                                 case hintExcisesFile:
    2498            2 :                                         // Account for the original file being deleted.
    2499            2 :                                         filesDeletedByCurrentHint++
    2500            2 :                                         // An excise could produce up to 2 new files. If the hint
    2501            2 :                                         // leaves a fragment of the file on the left, decrement
    2502            2 :                                         // the counter once. If the hint leaves a fragment of the
    2503            2 :                                         // file on the right, decrement the counter once.
    2504            2 :                                         if cmp(h.start, m.Smallest().UserKey) > 0 {
    2505            2 :                                                 filesDeletedByCurrentHint--
    2506            2 :                                         }
    2507            2 :                                         if m.UserKeyBounds().End.IsUpperBoundFor(cmp, h.end) {
    2508            2 :                                                 filesDeletedByCurrentHint--
    2509            2 :                                         }
    2510              :                                 }
    2511            2 :                                 filesDeletedByLevel[l] = append(filesDeletedByLevel[l], m)
    2512              :                         }
    2513              :                 }
    2514            2 :                 if filesDeletedByCurrentHint < 0 {
    2515            2 :                         // This hint does not delete a sufficient number of files to warrant
    2516            2 :                         // a delete-only compaction at this stage. Drop it (ie. don't add it
    2517            2 :                         // to either resolved or unresolved hints) so it doesn't stick around
    2518            2 :                         // forever.
    2519            2 :                         continue
    2520              :                 }
    2521              :                 // This hint will be resolved and dropped.
    2522            2 :                 for l := h.tombstoneLevel + 1; l < numLevels; l++ {
    2523            2 :                         byLevel[l] = append(byLevel[l], filesDeletedByLevel[l]...)
    2524            2 :                         for _, m := range filesDeletedByLevel[l] {
    2525            2 :                                 if files == nil {
    2526            2 :                                         // Construct files lazily, assuming most calls will not
    2527            2 :                                         // produce delete-only compactions.
    2528            2 :                                         files = make(map[*manifest.TableMetadata]bool)
    2529            2 :                                 }
    2530            2 :                                 files[m] = true
    2531              :                         }
    2532              :                 }
    2533            2 :                 resolvedHints = append(resolvedHints, h)
    2534              :         }
    2535              : 
    2536            2 :         var compactLevels []compactionLevel
    2537            2 :         for l, files := range byLevel {
    2538            2 :                 if len(files) == 0 {
    2539            2 :                         continue
    2540              :                 }
    2541            2 :                 compactLevels = append(compactLevels, compactionLevel{
    2542            2 :                         level: l,
    2543            2 :                         files: manifest.NewLevelSliceKeySorted(cmp, files),
    2544            2 :                 })
    2545              :         }
    2546            2 :         return compactLevels, resolvedHints, unresolvedHints
    2547              : }
    2548              : 
    2549              : // compact runs one compaction and maybe schedules another call to compact.
    2550            2 : func (d *DB) compact(c compaction, errChannel chan error) {
    2551            2 :         pprof.Do(context.Background(), c.PprofLabels(d.opts.Experimental.UserKeyCategories), func(context.Context) {
    2552            2 :                 func() {
    2553            2 :                         d.mu.Lock()
    2554            2 :                         defer d.mu.Unlock()
    2555            2 :                         jobID := d.newJobIDLocked()
    2556            2 : 
    2557            2 :                         compactErr := c.Execute(jobID, d)
    2558            2 : 
    2559            2 :                         d.deleteObsoleteFiles(jobID)
    2560            2 :                         // We send on the error channel only after we've deleted
    2561            2 :                         // obsolete files so that tests performing manual compactions
    2562            2 :                         // block until the obsolete files are deleted, and the test
    2563            2 :                         // observes the deletion.
    2564            2 :                         if errChannel != nil {
    2565            2 :                                 errChannel <- compactErr
    2566            2 :                         }
    2567            2 :                         if compactErr != nil {
    2568            2 :                                 d.handleCompactFailure(c, compactErr)
    2569            2 :                         }
    2570            2 :                         if c.IsDownload() {
    2571            2 :                                 d.mu.compact.downloadingCount--
    2572            2 :                         } else {
    2573            2 :                                 d.mu.compact.compactingCount--
    2574            2 :                         }
    2575            2 :                         delete(d.mu.compact.inProgress, c)
    2576            2 :                         // Add this compaction's duration to the cumulative duration. NB: This
    2577            2 :                         // must be atomic with the above removal of c from
    2578            2 :                         // d.mu.compact.InProgress to ensure Metrics.Compact.Duration does not
    2579            2 :                         // miss or double count a completing compaction's duration.
    2580            2 :                         d.mu.compact.duration += d.timeNow().Sub(c.BeganAt())
    2581              :                 }()
    2582              :                 // Done must not be called while holding any lock that needs to be
    2583              :                 // acquired by Schedule. Also, it must be called after new Version has
    2584              :                 // been installed, and metadata related to compactingCount and inProgress
    2585              :                 // compactions has been updated. This is because when we are running at
    2586              :                 // the limit of permitted compactions, Done can cause the
    2587              :                 // CompactionScheduler to schedule another compaction. Note that the only
    2588              :                 // compactions that may be scheduled by Done are those integrated with the
    2589              :                 // CompactionScheduler.
    2590            2 :                 c.GrantHandle().Done()
    2591            2 :                 // The previous compaction may have produced too many files in a level, so
    2592            2 :                 // reschedule another compaction if needed.
    2593            2 :                 //
    2594            2 :                 // The preceding Done call will not necessarily cause a compaction to be
    2595            2 :                 // scheduled, so we also need to call maybeScheduleCompaction. And
    2596            2 :                 // maybeScheduleCompaction encompasses all compactions, and not only those
    2597            2 :                 // scheduled via the CompactionScheduler.
    2598            2 :                 func() {
    2599            2 :                         d.mu.Lock()
    2600            2 :                         defer d.mu.Unlock()
    2601            2 :                         d.maybeScheduleCompaction()
    2602            2 :                         d.mu.compact.cond.Broadcast()
    2603            2 :                 }()
    2604              :         })
    2605              : }
    2606              : 
    2607            2 : func (d *DB) handleCompactFailure(c compaction, err error) {
    2608            2 :         if errors.Is(err, ErrCancelledCompaction) {
    2609            2 :                 // ErrCancelledCompaction is expected during normal operation, so we don't
    2610            2 :                 // want to report it as a background error.
    2611            2 :                 d.opts.Logger.Infof("%v", err)
    2612            2 :                 return
    2613            2 :         }
    2614            1 :         c.RecordError(&d.problemSpans, err)
    2615            1 :         // TODO(peter): count consecutive compaction errors and backoff.
    2616            1 :         d.opts.EventListener.BackgroundError(err)
    2617              : }
    2618              : 
    2619              : // cleanupVersionEdit cleans up any on-disk artifacts that were created
    2620              : // for the application of a versionEdit that is no longer going to be applied.
    2621              : //
    2622              : // d.mu must be held when calling this method.
    2623            2 : func (d *DB) cleanupVersionEdit(ve *manifest.VersionEdit) {
    2624            2 :         obsoleteFiles := manifest.ObsoleteFiles{
    2625            2 :                 TableBackings: make([]*manifest.TableBacking, 0, len(ve.NewTables)),
    2626            2 :                 BlobFiles:     make([]*manifest.PhysicalBlobFile, 0, len(ve.NewBlobFiles)),
    2627            2 :         }
    2628            2 :         deletedTables := make(map[base.TableNum]struct{})
    2629            2 :         for key := range ve.DeletedTables {
    2630            2 :                 deletedTables[key.FileNum] = struct{}{}
    2631            2 :         }
    2632            2 :         for i := range ve.NewBlobFiles {
    2633            1 :                 obsoleteFiles.AddBlob(ve.NewBlobFiles[i].Physical)
    2634            1 :                 d.mu.versions.zombieBlobs.Add(objectInfo{
    2635            1 :                         fileInfo: fileInfo{
    2636            1 :                                 FileNum:  ve.NewBlobFiles[i].Physical.FileNum,
    2637            1 :                                 FileSize: ve.NewBlobFiles[i].Physical.Size,
    2638            1 :                         },
    2639            1 :                         isLocal: objstorage.IsLocalBlobFile(d.objProvider, ve.NewBlobFiles[i].Physical.FileNum),
    2640            1 :                 })
    2641            1 :         }
    2642            2 :         for i := range ve.NewTables {
    2643            2 :                 if ve.NewTables[i].Meta.Virtual {
    2644            1 :                         // We handle backing files separately.
    2645            1 :                         continue
    2646              :                 }
    2647            2 :                 if _, ok := deletedTables[ve.NewTables[i].Meta.TableNum]; ok {
    2648            1 :                         // This file is being moved in this ve to a different level.
    2649            1 :                         // Don't mark it as obsolete.
    2650            1 :                         continue
    2651              :                 }
    2652            2 :                 obsoleteFiles.AddBacking(ve.NewTables[i].Meta.PhysicalMeta().TableBacking)
    2653              :         }
    2654            2 :         for i := range ve.CreatedBackingTables {
    2655            1 :                 if ve.CreatedBackingTables[i].IsUnused() {
    2656            0 :                         obsoleteFiles.AddBacking(ve.CreatedBackingTables[i])
    2657            0 :                 }
    2658              :         }
    2659            2 :         for _, of := range obsoleteFiles.TableBackings {
    2660            2 :                 // Add this file to zombie tables as well, as the versionSet
    2661            2 :                 // asserts on whether every obsolete file was at one point
    2662            2 :                 // marked zombie.
    2663            2 :                 d.mu.versions.zombieTables.Add(objectInfo{
    2664            2 :                         fileInfo: fileInfo{
    2665            2 :                                 FileNum:  of.DiskFileNum,
    2666            2 :                                 FileSize: of.Size,
    2667            2 :                         },
    2668            2 :                         isLocal: objstorage.IsLocalTable(d.objProvider, of.DiskFileNum),
    2669            2 :                 })
    2670            2 :         }
    2671            2 :         d.mu.versions.addObsoleteLocked(obsoleteFiles)
    2672              : }
    2673              : 
    2674              : // compact1 runs one compaction.
    2675              : //
    2676              : // d.mu must be held when calling this, but the mutex may be dropped and
    2677              : // re-acquired during the course of this method.
    2678            2 : func (d *DB) compact1(jobID JobID, c *tableCompaction) (err error) {
    2679            2 :         info := c.makeInfo(jobID)
    2680            2 :         d.opts.EventListener.CompactionBegin(info)
    2681            2 :         startTime := d.timeNow()
    2682            2 : 
    2683            2 :         ve, stats, err := d.runCompaction(jobID, c)
    2684            2 : 
    2685            2 :         info.Duration = d.timeNow().Sub(startTime)
    2686            2 :         if err == nil {
    2687            2 :                 validateVersionEdit(ve, d.opts.Comparer.ValidateKey, d.opts.Comparer.FormatKey, d.opts.Logger)
    2688            2 :                 err = d.mu.versions.UpdateVersionLocked(func() (versionUpdate, error) {
    2689            2 :                         // Check if this compaction had a conflicting operation (eg. a d.excise())
    2690            2 :                         // that necessitates it restarting from scratch. Note that since we hold
    2691            2 :                         // the manifest lock, we don't expect this bool to change its value
    2692            2 :                         // as only the holder of the manifest lock will ever write to it.
    2693            2 :                         if c.cancel.Load() {
    2694            2 :                                 err = firstError(err, ErrCancelledCompaction)
    2695            2 :                                 // This is the first time we've seen a cancellation during the
    2696            2 :                                 // life of this compaction (or the original condition on err == nil
    2697            2 :                                 // would not have been true). We should delete any tables already
    2698            2 :                                 // created, as d.runCompaction did not do that.
    2699            2 :                                 d.cleanupVersionEdit(ve)
    2700            2 :                                 // Note that UpdateVersionLocked invalidates the pickedCompactionCache
    2701            2 :                                 // when we return, which is relevant because this failed compaction
    2702            2 :                                 // may be the highest priority to run next.
    2703            2 :                                 return versionUpdate{}, err
    2704            2 :                         }
    2705            2 :                         return versionUpdate{
    2706            2 :                                 VE:                      ve,
    2707            2 :                                 JobID:                   jobID,
    2708            2 :                                 Metrics:                 c.metrics.perLevel,
    2709            2 :                                 InProgressCompactionsFn: func() []compactionInfo { return d.getInProgressCompactionInfoLocked(c) },
    2710              :                         }, nil
    2711              :                 })
    2712              :         }
    2713              : 
    2714            2 :         info.Done = true
    2715            2 :         info.Err = err
    2716            2 :         if err == nil {
    2717            2 :                 for i := range ve.NewTables {
    2718            2 :                         e := &ve.NewTables[i]
    2719            2 :                         info.Output.Tables = append(info.Output.Tables, e.Meta.TableInfo())
    2720            2 :                 }
    2721            2 :                 d.mu.snapshots.cumulativePinnedCount += stats.CumulativePinnedKeys
    2722            2 :                 d.mu.snapshots.cumulativePinnedSize += stats.CumulativePinnedSize
    2723            2 :                 d.mu.versions.metrics.Keys.MissizedTombstonesCount += stats.CountMissizedDels
    2724              :         }
    2725              : 
    2726              :         // NB: clearing compacting state must occur before updating the read state;
    2727              :         // L0Sublevels initialization depends on it.
    2728            2 :         d.clearCompactingState(c, err != nil)
    2729            2 :         d.mu.versions.incrementCompactions(c.kind, c.extraLevels, c.metrics.picker,
    2730            2 :                 c.metrics.bytesWritten.Load(), err)
    2731            2 :         d.mu.versions.incrementCompactionBytes(-c.metrics.bytesWritten.Load())
    2732            2 : 
    2733            2 :         info.TotalDuration = d.timeNow().Sub(c.metrics.beganAt)
    2734            2 :         d.opts.EventListener.CompactionEnd(info)
    2735            2 : 
    2736            2 :         // Update the read state before deleting obsolete files because the
    2737            2 :         // read-state update will cause the previous version to be unref'd and if
    2738            2 :         // there are no references obsolete tables will be added to the obsolete
    2739            2 :         // table list.
    2740            2 :         if err == nil {
    2741            2 :                 d.updateReadStateLocked(d.opts.DebugCheck)
    2742            2 :                 d.updateTableStatsLocked(ve.NewTables)
    2743            2 :         }
    2744              : 
    2745            2 :         return err
    2746              : }
    2747              : 
    2748              : // runCopyCompaction runs a copy compaction where a new TableNum is created that
    2749              : // is a byte-for-byte copy of the input file or span thereof in some cases. This
    2750              : // is used in lieu of a move compaction when a file is being moved across the
    2751              : // local/remote storage boundary. It could also be used in lieu of a rewrite
    2752              : // compaction as part of a Download() call, which allows copying only a span of
    2753              : // the external file, provided the file does not contain range keys or value
    2754              : // blocks (see sstable.CopySpan).
    2755              : //
    2756              : // d.mu must be held when calling this method. The mutex will be released when
    2757              : // doing IO.
    2758              : func (d *DB) runCopyCompaction(
    2759              :         jobID JobID, c *tableCompaction,
    2760            2 : ) (ve *manifest.VersionEdit, stats compact.Stats, _ error) {
    2761            2 :         if c.cancel.Load() {
    2762            0 :                 return nil, compact.Stats{}, ErrCancelledCompaction
    2763            0 :         }
    2764            2 :         iter := c.startLevel.files.Iter()
    2765            2 :         inputMeta := iter.First()
    2766            2 :         if iter.Next() != nil {
    2767            0 :                 return nil, compact.Stats{}, base.AssertionFailedf("got more than one file for a move compaction")
    2768            0 :         }
    2769            2 :         if inputMeta.BlobReferenceDepth > 0 || len(inputMeta.BlobReferences) > 0 {
    2770            0 :                 return nil, compact.Stats{}, base.AssertionFailedf(
    2771            0 :                         "copy compaction for %s with blob references (depth=%d, refs=%d)",
    2772            0 :                         inputMeta.TableNum, inputMeta.BlobReferenceDepth, len(inputMeta.BlobReferences),
    2773            0 :                 )
    2774            0 :         }
    2775            2 :         ve = &manifest.VersionEdit{
    2776            2 :                 DeletedTables: map[manifest.DeletedTableEntry]*manifest.TableMetadata{
    2777            2 :                         {Level: c.startLevel.level, FileNum: inputMeta.TableNum}: inputMeta,
    2778            2 :                 },
    2779            2 :         }
    2780            2 : 
    2781            2 :         objMeta, err := d.objProvider.Lookup(base.FileTypeTable, inputMeta.TableBacking.DiskFileNum)
    2782            2 :         if err != nil {
    2783            0 :                 return nil, compact.Stats{}, err
    2784            0 :         }
    2785              :         // This code does not support copying a shared table (which should never be necessary).
    2786            2 :         if objMeta.IsShared() {
    2787            0 :                 return nil, compact.Stats{}, base.AssertionFailedf("copy compaction of shared table")
    2788            0 :         }
    2789              : 
    2790              :         // We are in the relatively more complex case where we need to copy this
    2791              :         // file to remote storage. Drop the db mutex while we do the copy
    2792              :         //
    2793              :         // To ease up cleanup of the local file and tracking of refs, we create
    2794              :         // a new FileNum. This has the potential of making the block cache less
    2795              :         // effective, however.
    2796            2 :         newMeta := &manifest.TableMetadata{
    2797            2 :                 Size:                     inputMeta.Size,
    2798            2 :                 CreationTime:             inputMeta.CreationTime,
    2799            2 :                 SmallestSeqNum:           inputMeta.SmallestSeqNum,
    2800            2 :                 LargestSeqNum:            inputMeta.LargestSeqNum,
    2801            2 :                 LargestSeqNumAbsolute:    inputMeta.LargestSeqNumAbsolute,
    2802            2 :                 Stats:                    inputMeta.Stats,
    2803            2 :                 Virtual:                  inputMeta.Virtual,
    2804            2 :                 SyntheticPrefixAndSuffix: inputMeta.SyntheticPrefixAndSuffix,
    2805            2 :         }
    2806            2 :         if inputMeta.HasPointKeys {
    2807            2 :                 newMeta.ExtendPointKeyBounds(c.comparer.Compare,
    2808            2 :                         inputMeta.PointKeyBounds.Smallest(),
    2809            2 :                         inputMeta.PointKeyBounds.Largest())
    2810            2 :         }
    2811            2 :         if inputMeta.HasRangeKeys {
    2812            2 :                 newMeta.ExtendRangeKeyBounds(c.comparer.Compare,
    2813            2 :                         inputMeta.RangeKeyBounds.Smallest(),
    2814            2 :                         inputMeta.RangeKeyBounds.Largest())
    2815            2 :         }
    2816            2 :         newMeta.TableNum = d.mu.versions.getNextTableNum()
    2817            2 :         if objMeta.IsExternal() {
    2818            2 :                 // external -> local/shared copy. File must be virtual.
    2819            2 :                 // We will update this size later after we produce the new backing file.
    2820            2 :                 newMeta.InitVirtualBacking(base.DiskFileNum(newMeta.TableNum), inputMeta.TableBacking.Size)
    2821            2 :         } else {
    2822            2 :                 // local -> shared copy. New file is guaranteed to not be virtual.
    2823            2 :                 newMeta.InitPhysicalBacking()
    2824            2 :         }
    2825              : 
    2826              :         // NB: The order here is reversed, lock after unlock. This is similar to
    2827              :         // runCompaction.
    2828            2 :         d.mu.Unlock()
    2829            2 :         defer d.mu.Lock()
    2830            2 : 
    2831            2 :         deleteOnExit := false
    2832            2 :         defer func() {
    2833            2 :                 if deleteOnExit {
    2834            1 :                         _ = d.objProvider.Remove(base.FileTypeTable, newMeta.TableBacking.DiskFileNum)
    2835            1 :                 }
    2836              :         }()
    2837              : 
    2838              :         // If the src obj is external, we're doing an external to local/shared copy.
    2839            2 :         if objMeta.IsExternal() {
    2840            2 :                 ctx := context.TODO()
    2841            2 :                 src, err := d.objProvider.OpenForReading(
    2842            2 :                         ctx, base.FileTypeTable, inputMeta.TableBacking.DiskFileNum, objstorage.OpenOptions{},
    2843            2 :                 )
    2844            2 :                 if err != nil {
    2845            0 :                         return nil, compact.Stats{}, err
    2846            0 :                 }
    2847            2 :                 defer func() {
    2848            2 :                         if src != nil {
    2849            0 :                                 _ = src.Close()
    2850            0 :                         }
    2851              :                 }()
    2852              : 
    2853            2 :                 w, _, err := d.objProvider.Create(
    2854            2 :                         ctx, base.FileTypeTable, newMeta.TableBacking.DiskFileNum,
    2855            2 :                         objstorage.CreateOptions{
    2856            2 :                                 PreferSharedStorage: d.shouldCreateShared(c.outputLevel.level),
    2857            2 :                         },
    2858            2 :                 )
    2859            2 :                 if err != nil {
    2860            0 :                         return nil, compact.Stats{}, err
    2861            0 :                 }
    2862            2 :                 deleteOnExit = true
    2863            2 : 
    2864            2 :                 start, end := newMeta.Smallest(), newMeta.Largest()
    2865            2 :                 if newMeta.SyntheticPrefixAndSuffix.HasPrefix() {
    2866            2 :                         syntheticPrefix := newMeta.SyntheticPrefixAndSuffix.Prefix()
    2867            2 :                         start.UserKey = syntheticPrefix.Invert(start.UserKey)
    2868            2 :                         end.UserKey = syntheticPrefix.Invert(end.UserKey)
    2869            2 :                 }
    2870            2 :                 if newMeta.SyntheticPrefixAndSuffix.HasSuffix() {
    2871            1 :                         // Extend the bounds as necessary so that the keys don't include suffixes.
    2872            1 :                         start.UserKey = start.UserKey[:c.comparer.Split(start.UserKey)]
    2873            1 :                         if n := c.comparer.Split(end.UserKey); n < len(end.UserKey) {
    2874            0 :                                 end = base.MakeRangeDeleteSentinelKey(c.comparer.ImmediateSuccessor(nil, end.UserKey[:n]))
    2875            0 :                         }
    2876              :                 }
    2877              : 
    2878              :                 // NB: external files are always virtual.
    2879            2 :                 var wrote uint64
    2880            2 :                 err = d.fileCache.withReader(ctx, block.NoReadEnv, inputMeta.VirtualMeta(), func(r *sstable.Reader, env sstable.ReadEnv) error {
    2881            2 :                         var err error
    2882            2 :                         // TODO(radu): plumb a ReadEnv to CopySpan (it could use the buffer pool
    2883            2 :                         // or update category stats).
    2884            2 :                         wrote, err = sstable.CopySpan(ctx,
    2885            2 :                                 src, r, d.opts.MakeReaderOptions(),
    2886            2 :                                 w, d.opts.MakeWriterOptions(c.outputLevel.level, d.TableFormat()),
    2887            2 :                                 start, end,
    2888            2 :                         )
    2889            2 :                         return err
    2890            2 :                 })
    2891              : 
    2892            2 :                 src = nil // We passed src to CopySpan; it's responsible for closing it.
    2893            2 :                 if err != nil {
    2894            1 :                         if errors.Is(err, sstable.ErrEmptySpan) {
    2895            1 :                                 // The virtual table was empty. Just remove the backing file.
    2896            1 :                                 // Note that deleteOnExit is true so we will delete the created object.
    2897            1 :                                 outputMetrics := c.metrics.perLevel.level(c.outputLevel.level)
    2898            1 :                                 outputMetrics.TableBytesIn = inputMeta.Size
    2899            1 : 
    2900            1 :                                 return ve, compact.Stats{}, nil
    2901            1 :                         }
    2902            0 :                         return nil, compact.Stats{}, err
    2903              :                 }
    2904            2 :                 newMeta.TableBacking.Size = wrote
    2905            2 :                 newMeta.Size = wrote
    2906            2 :         } else {
    2907            2 :                 _, err := d.objProvider.LinkOrCopyFromLocal(context.TODO(), d.opts.FS,
    2908            2 :                         d.objProvider.Path(objMeta), base.FileTypeTable, newMeta.TableBacking.DiskFileNum,
    2909            2 :                         objstorage.CreateOptions{PreferSharedStorage: true})
    2910            2 :                 if err != nil {
    2911            0 :                         return nil, compact.Stats{}, err
    2912            0 :                 }
    2913            2 :                 deleteOnExit = true
    2914              :         }
    2915            2 :         ve.NewTables = []manifest.NewTableEntry{{
    2916            2 :                 Level: c.outputLevel.level,
    2917            2 :                 Meta:  newMeta,
    2918            2 :         }}
    2919            2 :         if newMeta.Virtual {
    2920            2 :                 ve.CreatedBackingTables = []*manifest.TableBacking{newMeta.TableBacking}
    2921            2 :         }
    2922            2 :         outputMetrics := c.metrics.perLevel.level(c.outputLevel.level)
    2923            2 :         outputMetrics.TableBytesIn = inputMeta.Size
    2924            2 :         outputMetrics.TableBytesCompacted = newMeta.Size
    2925            2 :         outputMetrics.TablesCompacted = 1
    2926            2 : 
    2927            2 :         if err := d.objProvider.Sync(); err != nil {
    2928            0 :                 return nil, compact.Stats{}, err
    2929            0 :         }
    2930            2 :         deleteOnExit = false
    2931            2 :         return ve, compact.Stats{}, nil
    2932              : }
    2933              : 
    2934              : // applyHintOnFile applies a deleteCompactionHint to a file, and updates the
    2935              : // versionEdit accordingly. It returns a list of new files that were created
    2936              : // if the hint was applied partially to a file (eg. through an exciseTable as opposed
    2937              : // to an outright deletion). levelMetrics is kept up-to-date with the number
    2938              : // of tables deleted or excised.
    2939              : func (d *DB) applyHintOnFile(
    2940              :         h deleteCompactionHint,
    2941              :         f *manifest.TableMetadata,
    2942              :         level int,
    2943              :         levelMetrics *LevelMetrics,
    2944              :         ve *manifest.VersionEdit,
    2945              :         hintOverlap deletionHintOverlap,
    2946            2 : ) (newFiles []manifest.NewTableEntry, err error) {
    2947            2 :         if hintOverlap == hintDoesNotApply {
    2948            0 :                 return nil, nil
    2949            0 :         }
    2950              : 
    2951              :         // The hint overlaps with at least part of the file.
    2952            2 :         if hintOverlap == hintDeletesFile {
    2953            2 :                 // The hint deletes the entirety of this file.
    2954            2 :                 ve.DeletedTables[manifest.DeletedTableEntry{
    2955            2 :                         Level:   level,
    2956            2 :                         FileNum: f.TableNum,
    2957            2 :                 }] = f
    2958            2 :                 levelMetrics.TablesDeleted++
    2959            2 :                 return nil, nil
    2960            2 :         }
    2961              :         // The hint overlaps with only a part of the file, not the entirety of it. We need
    2962              :         // to use d.exciseTable. (hintOverlap == hintExcisesFile)
    2963            2 :         if d.FormatMajorVersion() < FormatVirtualSSTables {
    2964            0 :                 panic("pebble: delete-only compaction hint excising a file is not supported in this version")
    2965              :         }
    2966              : 
    2967            2 :         levelMetrics.TablesExcised++
    2968            2 :         exciseBounds := base.UserKeyBoundsEndExclusive(h.start, h.end)
    2969            2 :         leftTable, rightTable, err := d.exciseTable(context.TODO(), exciseBounds, f, level, tightExciseBounds)
    2970            2 :         if err != nil {
    2971            0 :                 return nil, errors.Wrap(err, "error when running excise for delete-only compaction")
    2972            0 :         }
    2973            2 :         newFiles = applyExciseToVersionEdit(ve, f, leftTable, rightTable, level)
    2974            2 :         return newFiles, nil
    2975              : }
    2976              : 
    2977              : func (d *DB) runDeleteOnlyCompactionForLevel(
    2978              :         cl compactionLevel,
    2979              :         levelMetrics *LevelMetrics,
    2980              :         ve *manifest.VersionEdit,
    2981              :         snapshots compact.Snapshots,
    2982              :         fragments []deleteCompactionHintFragment,
    2983              :         exciseEnabled bool,
    2984            2 : ) error {
    2985            2 :         if cl.level == 0 {
    2986            0 :                 panic("cannot run delete-only compaction for L0")
    2987              :         }
    2988            2 :         curFragment := 0
    2989            2 : 
    2990            2 :         // Outer loop loops on files. Middle loop loops on fragments. Inner loop
    2991            2 :         // loops on raw fragments of hints. Number of fragments are bounded by
    2992            2 :         // the number of hints this compaction was created with, which is capped
    2993            2 :         // in the compaction picker to avoid very CPU-hot loops here.
    2994            2 :         for f := range cl.files.All() {
    2995            2 :                 // curFile usually matches f, except if f got excised in which case
    2996            2 :                 // it maps to a virtual file that replaces f, or nil if f got removed
    2997            2 :                 // in its entirety.
    2998            2 :                 curFile := f
    2999            2 :                 for curFragment < len(fragments) && d.cmp(fragments[curFragment].start, f.Smallest().UserKey) <= 0 {
    3000            2 :                         curFragment++
    3001            2 :                 }
    3002            2 :                 if curFragment > 0 {
    3003            2 :                         curFragment--
    3004            2 :                 }
    3005              : 
    3006            2 :                 for ; curFragment < len(fragments); curFragment++ {
    3007            2 :                         if f.UserKeyBounds().End.CompareUpperBounds(d.cmp, base.UserKeyInclusive(fragments[curFragment].start)) < 0 {
    3008            2 :                                 break
    3009              :                         }
    3010              :                         // Process all overlapping hints with this file. Note that applying
    3011              :                         // a hint twice is idempotent; curFile should have already been excised
    3012              :                         // the first time, resulting in no change the second time.
    3013            2 :                         for _, h := range fragments[curFragment].hints {
    3014            2 :                                 if h.tombstoneLevel >= cl.level {
    3015            2 :                                         // We cannot excise out the deletion tombstone itself, or anything
    3016            2 :                                         // above it.
    3017            2 :                                         continue
    3018              :                                 }
    3019            2 :                                 hintOverlap := h.canDeleteOrExcise(d.cmp, curFile, snapshots, exciseEnabled)
    3020            2 :                                 if hintOverlap == hintDoesNotApply {
    3021            2 :                                         continue
    3022              :                                 }
    3023            2 :                                 newFiles, err := d.applyHintOnFile(h, curFile, cl.level, levelMetrics, ve, hintOverlap)
    3024            2 :                                 if err != nil {
    3025            0 :                                         return err
    3026            0 :                                 }
    3027            2 :                                 if _, ok := ve.DeletedTables[manifest.DeletedTableEntry{Level: cl.level, FileNum: curFile.TableNum}]; ok {
    3028            2 :                                         curFile = nil
    3029            2 :                                 }
    3030            2 :                                 if len(newFiles) > 0 {
    3031            2 :                                         curFile = newFiles[len(newFiles)-1].Meta
    3032            2 :                                 } else if curFile == nil {
    3033            2 :                                         // Nothing remains of the file.
    3034            2 :                                         break
    3035              :                                 }
    3036              :                         }
    3037            2 :                         if curFile == nil {
    3038            2 :                                 // Nothing remains of the file.
    3039            2 :                                 break
    3040              :                         }
    3041              :                 }
    3042            2 :                 if _, ok := ve.DeletedTables[manifest.DeletedTableEntry{
    3043            2 :                         Level:   cl.level,
    3044            2 :                         FileNum: f.TableNum,
    3045            2 :                 }]; !ok {
    3046            0 :                         panic("pebble: delete-only compaction scheduled with hints that did not delete or excise a file")
    3047              :                 }
    3048              :         }
    3049            2 :         return nil
    3050              : }
    3051              : 
    3052              : // deleteCompactionHintFragment represents a fragment of the key space and
    3053              : // contains a set of deleteCompactionHints that apply to that fragment; a
    3054              : // fragment starts at the start field and ends where the next fragment starts.
    3055              : type deleteCompactionHintFragment struct {
    3056              :         start []byte
    3057              :         hints []deleteCompactionHint
    3058              : }
    3059              : 
    3060              : // Delete compaction hints can overlap with each other, and multiple fragments
    3061              : // can apply to a single file. This function takes a list of hints and fragments
    3062              : // them, to make it easier to apply them to non-overlapping files occupying a level;
    3063              : // that way, files and hint fragments can be iterated on in lockstep, while efficiently
    3064              : // being able to apply all hints overlapping with a given file.
    3065              : func fragmentDeleteCompactionHints(
    3066              :         cmp Compare, hints []deleteCompactionHint,
    3067            2 : ) []deleteCompactionHintFragment {
    3068            2 :         fragments := make([]deleteCompactionHintFragment, 0, len(hints)*2)
    3069            2 :         for i := range hints {
    3070            2 :                 fragments = append(fragments, deleteCompactionHintFragment{start: hints[i].start},
    3071            2 :                         deleteCompactionHintFragment{start: hints[i].end})
    3072            2 :         }
    3073            2 :         slices.SortFunc(fragments, func(i, j deleteCompactionHintFragment) int {
    3074            2 :                 return cmp(i.start, j.start)
    3075            2 :         })
    3076            2 :         fragments = slices.CompactFunc(fragments, func(i, j deleteCompactionHintFragment) bool {
    3077            2 :                 return bytes.Equal(i.start, j.start)
    3078            2 :         })
    3079            2 :         for _, h := range hints {
    3080            2 :                 startIdx := sort.Search(len(fragments), func(i int) bool {
    3081            2 :                         return cmp(fragments[i].start, h.start) >= 0
    3082            2 :                 })
    3083            2 :                 endIdx := sort.Search(len(fragments), func(i int) bool {
    3084            2 :                         return cmp(fragments[i].start, h.end) >= 0
    3085            2 :                 })
    3086            2 :                 for i := startIdx; i < endIdx; i++ {
    3087            2 :                         fragments[i].hints = append(fragments[i].hints, h)
    3088            2 :                 }
    3089              :         }
    3090            2 :         return fragments
    3091              : }
    3092              : 
    3093              : // Runs a delete-only compaction.
    3094              : //
    3095              : // d.mu must *not* be held when calling this.
    3096              : func (d *DB) runDeleteOnlyCompaction(
    3097              :         jobID JobID, c *tableCompaction, snapshots compact.Snapshots,
    3098            2 : ) (ve *manifest.VersionEdit, stats compact.Stats, retErr error) {
    3099            2 :         fragments := fragmentDeleteCompactionHints(d.cmp, c.deleteOnly.hints)
    3100            2 :         ve = &manifest.VersionEdit{
    3101            2 :                 DeletedTables: map[manifest.DeletedTableEntry]*manifest.TableMetadata{},
    3102            2 :         }
    3103            2 :         for _, cl := range c.inputs {
    3104            2 :                 levelMetrics := c.metrics.perLevel.level(cl.level)
    3105            2 :                 err := d.runDeleteOnlyCompactionForLevel(cl, levelMetrics, ve, snapshots, fragments, c.deleteOnly.exciseEnabled)
    3106            2 :                 if err != nil {
    3107            0 :                         return nil, stats, err
    3108            0 :                 }
    3109              :         }
    3110              :         // Remove any files that were added and deleted in the same versionEdit.
    3111            2 :         ve.NewTables = slices.DeleteFunc(ve.NewTables, func(e manifest.NewTableEntry) bool {
    3112            2 :                 entry := manifest.DeletedTableEntry{Level: e.Level, FileNum: e.Meta.TableNum}
    3113            2 :                 if _, deleted := ve.DeletedTables[entry]; deleted {
    3114            2 :                         delete(ve.DeletedTables, entry)
    3115            2 :                         return true
    3116            2 :                 }
    3117            2 :                 return false
    3118              :         })
    3119              :         // Remove any entries from CreatedBackingTables that are not used in any
    3120              :         // NewFiles.
    3121            2 :         usedBackingFiles := make(map[base.DiskFileNum]struct{})
    3122            2 :         for _, e := range ve.NewTables {
    3123            2 :                 if e.Meta.Virtual {
    3124            2 :                         usedBackingFiles[e.Meta.TableBacking.DiskFileNum] = struct{}{}
    3125            2 :                 }
    3126              :         }
    3127            2 :         ve.CreatedBackingTables = slices.DeleteFunc(ve.CreatedBackingTables, func(b *manifest.TableBacking) bool {
    3128            2 :                 _, used := usedBackingFiles[b.DiskFileNum]
    3129            2 :                 return !used
    3130            2 :         })
    3131              :         // Refresh the disk available statistic whenever a compaction/flush
    3132              :         // completes, before re-acquiring the mutex.
    3133            2 :         d.calculateDiskAvailableBytes()
    3134            2 :         return ve, stats, nil
    3135              : }
    3136              : 
    3137              : func (d *DB) runMoveCompaction(
    3138              :         jobID JobID, c *tableCompaction,
    3139            2 : ) (ve *manifest.VersionEdit, stats compact.Stats, _ error) {
    3140            2 :         iter := c.startLevel.files.Iter()
    3141            2 :         meta := iter.First()
    3142            2 :         if iter.Next() != nil {
    3143            0 :                 return nil, stats, base.AssertionFailedf("got more than one file for a move compaction")
    3144            0 :         }
    3145            2 :         if c.cancel.Load() {
    3146            0 :                 return ve, stats, ErrCancelledCompaction
    3147            0 :         }
    3148            2 :         outputMetrics := c.metrics.perLevel.level(c.outputLevel.level)
    3149            2 :         outputMetrics.TableBytesMoved = meta.Size
    3150            2 :         outputMetrics.TablesMoved = 1
    3151            2 :         ve = &manifest.VersionEdit{
    3152            2 :                 DeletedTables: map[manifest.DeletedTableEntry]*manifest.TableMetadata{
    3153            2 :                         {Level: c.startLevel.level, FileNum: meta.TableNum}: meta,
    3154            2 :                 },
    3155            2 :                 NewTables: []manifest.NewTableEntry{
    3156            2 :                         {Level: c.outputLevel.level, Meta: meta},
    3157            2 :                 },
    3158            2 :         }
    3159            2 : 
    3160            2 :         return ve, stats, nil
    3161              : }
    3162              : 
    3163              : // runCompaction runs a compaction that produces new on-disk tables from
    3164              : // memtables or old on-disk tables.
    3165              : //
    3166              : // runCompaction cannot be used for compactionKindIngestedFlushable.
    3167              : //
    3168              : // d.mu must be held when calling this, but the mutex may be dropped and
    3169              : // re-acquired during the course of this method.
    3170              : func (d *DB) runCompaction(
    3171              :         jobID JobID, c *tableCompaction,
    3172            2 : ) (ve *manifest.VersionEdit, stats compact.Stats, retErr error) {
    3173            2 :         if c.cancel.Load() {
    3174            2 :                 return ve, stats, ErrCancelledCompaction
    3175            2 :         }
    3176            2 :         switch c.kind {
    3177            2 :         case compactionKindDeleteOnly:
    3178            2 :                 // Release the d.mu lock while doing I/O.
    3179            2 :                 // Note the unusual order: Unlock and then Lock.
    3180            2 :                 snapshots := d.mu.snapshots.toSlice()
    3181            2 :                 d.mu.Unlock()
    3182            2 :                 defer d.mu.Lock()
    3183            2 :                 return d.runDeleteOnlyCompaction(jobID, c, snapshots)
    3184            2 :         case compactionKindMove:
    3185            2 :                 return d.runMoveCompaction(jobID, c)
    3186            2 :         case compactionKindCopy:
    3187            2 :                 return d.runCopyCompaction(jobID, c)
    3188            0 :         case compactionKindIngestedFlushable:
    3189            0 :                 panic("pebble: runCompaction cannot handle compactionKindIngestedFlushable.")
    3190              :         }
    3191              : 
    3192            2 :         snapshots := d.mu.snapshots.toSlice()
    3193            2 : 
    3194            2 :         // Release the d.mu lock while doing I/O.
    3195            2 :         // Note the unusual order: Unlock and then Lock.
    3196            2 :         d.mu.Unlock()
    3197            2 :         defer d.mu.Lock()
    3198            2 : 
    3199            2 :         // Determine whether we should separate values into blob files.
    3200            2 :         //
    3201            2 :         // TODO(jackson): Currently we never separate values in non-tests. Choose
    3202            2 :         // and initialize the appropriate ValueSeparation implementation based on
    3203            2 :         // Options and the compaction inputs.
    3204            2 :         valueSeparation := c.getValueSeparation(jobID, c, c.tableFormat)
    3205            2 : 
    3206            2 :         result := d.compactAndWrite(jobID, c, snapshots, c.tableFormat, valueSeparation)
    3207            2 :         if result.Err == nil {
    3208            2 :                 ve, result.Err = c.makeVersionEdit(result)
    3209            2 :         }
    3210            2 :         if result.Err != nil {
    3211            2 :                 // Delete any created tables or blob files.
    3212            2 :                 obsoleteFiles := manifest.ObsoleteFiles{
    3213            2 :                         TableBackings: make([]*manifest.TableBacking, 0, len(result.Tables)),
    3214            2 :                         BlobFiles:     make([]*manifest.PhysicalBlobFile, 0, len(result.Blobs)),
    3215            2 :                 }
    3216            2 :                 d.mu.Lock()
    3217            2 :                 for i := range result.Tables {
    3218            1 :                         backing := &manifest.TableBacking{
    3219            1 :                                 DiskFileNum: result.Tables[i].ObjMeta.DiskFileNum,
    3220            1 :                                 Size:        result.Tables[i].WriterMeta.Size,
    3221            1 :                         }
    3222            1 :                         obsoleteFiles.AddBacking(backing)
    3223            1 :                         // Add this file to zombie tables as well, as the versionSet
    3224            1 :                         // asserts on whether every obsolete file was at one point
    3225            1 :                         // marked zombie.
    3226            1 :                         d.mu.versions.zombieTables.AddMetadata(&result.Tables[i].ObjMeta, backing.Size)
    3227            1 :                 }
    3228            2 :                 for i := range result.Blobs {
    3229            0 :                         obsoleteFiles.AddBlob(result.Blobs[i].Metadata)
    3230            0 :                         // Add this file to zombie blobs as well, as the versionSet
    3231            0 :                         // asserts on whether every obsolete file was at one point
    3232            0 :                         // marked zombie.
    3233            0 :                         d.mu.versions.zombieBlobs.AddMetadata(&result.Blobs[i].ObjMeta, result.Blobs[i].Metadata.Size)
    3234            0 :                 }
    3235            2 :                 d.mu.versions.addObsoleteLocked(obsoleteFiles)
    3236            2 :                 d.mu.Unlock()
    3237              :         }
    3238              :         // Refresh the disk available statistic whenever a compaction/flush
    3239              :         // completes, before re-acquiring the mutex.
    3240            2 :         d.calculateDiskAvailableBytes()
    3241            2 :         return ve, result.Stats, result.Err
    3242              : }
    3243              : 
    3244              : // compactAndWrite runs the data part of a compaction, where we set up a
    3245              : // compaction iterator and use it to write output tables.
    3246              : func (d *DB) compactAndWrite(
    3247              :         jobID JobID,
    3248              :         c *tableCompaction,
    3249              :         snapshots compact.Snapshots,
    3250              :         tableFormat sstable.TableFormat,
    3251              :         valueSeparation compact.ValueSeparation,
    3252            2 : ) (result compact.Result) {
    3253            2 :         // Compactions use a pool of buffers to read blocks, avoiding polluting the
    3254            2 :         // block cache with blocks that will not be read again. We initialize the
    3255            2 :         // buffer pool with a size 12. This initial size does not need to be
    3256            2 :         // accurate, because the pool will grow to accommodate the maximum number of
    3257            2 :         // blocks allocated at a given time over the course of the compaction. But
    3258            2 :         // choosing a size larger than that working set avoids any additional
    3259            2 :         // allocations to grow the size of the pool over the course of iteration.
    3260            2 :         //
    3261            2 :         // Justification for initial size 12: In a two-level compaction, at any
    3262            2 :         // given moment we'll have 2 index blocks in-use and 2 data blocks in-use.
    3263            2 :         // Additionally, when decoding a compressed block, we'll temporarily
    3264            2 :         // allocate 1 additional block to hold the compressed buffer. In the worst
    3265            2 :         // case that all input sstables have two-level index blocks (+2), value
    3266            2 :         // blocks (+2), range deletion blocks (+n) and range key blocks (+n), we'll
    3267            2 :         // additionally require 2n+4 blocks where n is the number of input sstables.
    3268            2 :         // Range deletion and range key blocks are relatively rare, and the cost of
    3269            2 :         // an additional allocation or two over the course of the compaction is
    3270            2 :         // considered to be okay. A larger initial size would cause the pool to hold
    3271            2 :         // on to more memory, even when it's not in-use because the pool will
    3272            2 :         // recycle buffers up to the current capacity of the pool. The memory use of
    3273            2 :         // a 12-buffer pool is expected to be within reason, even if all the buffers
    3274            2 :         // grow to the typical size of an index block (256 KiB) which would
    3275            2 :         // translate to 3 MiB per compaction.
    3276            2 :         c.iterationState.bufferPool.Init(12)
    3277            2 :         defer c.iterationState.bufferPool.Release()
    3278            2 :         blockReadEnv := block.ReadEnv{
    3279            2 :                 BufferPool: &c.iterationState.bufferPool,
    3280            2 :                 Stats:      &c.metrics.internalIterStats,
    3281            2 :                 IterStats: d.fileCache.SSTStatsCollector().Accumulator(
    3282            2 :                         uint64(uintptr(unsafe.Pointer(c))),
    3283            2 :                         categoryCompaction,
    3284            2 :                 ),
    3285            2 :         }
    3286            2 :         if c.version != nil {
    3287            2 :                 c.iterationState.valueFetcher.Init(&c.version.BlobFiles, d.fileCache, blockReadEnv)
    3288            2 :         }
    3289            2 :         iiopts := internalIterOpts{
    3290            2 :                 compaction:       true,
    3291            2 :                 readEnv:          sstable.ReadEnv{Block: blockReadEnv},
    3292            2 :                 blobValueFetcher: &c.iterationState.valueFetcher,
    3293            2 :         }
    3294            2 :         defer func() { _ = c.iterationState.valueFetcher.Close() }()
    3295              : 
    3296            2 :         pointIter, rangeDelIter, rangeKeyIter, err := c.newInputIters(d.newIters, iiopts)
    3297            2 :         defer func() {
    3298            2 :                 for _, closer := range c.iterationState.keyspanIterClosers {
    3299            2 :                         closer.FragmentIterator.Close()
    3300            2 :                 }
    3301              :         }()
    3302            2 :         if err != nil {
    3303            1 :                 return compact.Result{Err: err}
    3304            1 :         }
    3305            2 :         cfg := compact.IterConfig{
    3306            2 :                 Comparer:         c.comparer,
    3307            2 :                 Merge:            d.merge,
    3308            2 :                 TombstoneElision: c.delElision,
    3309            2 :                 RangeKeyElision:  c.rangeKeyElision,
    3310            2 :                 Snapshots:        snapshots,
    3311            2 :                 AllowZeroSeqNum:  c.allowZeroSeqNum(),
    3312            2 :                 IneffectualSingleDeleteCallback: func(userKey []byte) {
    3313            2 :                         d.opts.EventListener.PossibleAPIMisuse(PossibleAPIMisuseInfo{
    3314            2 :                                 Kind:    IneffectualSingleDelete,
    3315            2 :                                 UserKey: slices.Clone(userKey),
    3316            2 :                         })
    3317            2 :                 },
    3318            0 :                 NondeterministicSingleDeleteCallback: func(userKey []byte) {
    3319            0 :                         d.opts.EventListener.PossibleAPIMisuse(PossibleAPIMisuseInfo{
    3320            0 :                                 Kind:    NondeterministicSingleDelete,
    3321            0 :                                 UserKey: slices.Clone(userKey),
    3322            0 :                         })
    3323            0 :                 },
    3324            2 :                 MissizedDeleteCallback: func(userKey []byte, elidedSize, expectedSize uint64) {
    3325            2 :                         d.opts.EventListener.PossibleAPIMisuse(PossibleAPIMisuseInfo{
    3326            2 :                                 Kind:    MissizedDelete,
    3327            2 :                                 UserKey: slices.Clone(userKey),
    3328            2 :                                 ExtraInfo: redact.Sprintf("elidedSize=%d,expectedSize=%d",
    3329            2 :                                         redact.SafeUint(elidedSize), redact.SafeUint(expectedSize)),
    3330            2 :                         })
    3331            2 :                 },
    3332              :         }
    3333            2 :         iter := compact.NewIter(cfg, pointIter, rangeDelIter, rangeKeyIter)
    3334            2 : 
    3335            2 :         runnerCfg := compact.RunnerConfig{
    3336            2 :                 CompactionBounds:           c.bounds,
    3337            2 :                 L0SplitKeys:                c.flush.l0Limits,
    3338            2 :                 Grandparents:               c.grandparents,
    3339            2 :                 MaxGrandparentOverlapBytes: c.maxOverlapBytes,
    3340            2 :                 TargetOutputFileSize:       c.maxOutputFileSize,
    3341            2 :                 GrantHandle:                c.grantHandle,
    3342            2 :         }
    3343            2 :         runner := compact.NewRunner(runnerCfg, iter)
    3344            2 : 
    3345            2 :         var spanPolicyValid bool
    3346            2 :         var spanPolicy SpanPolicy
    3347            2 :         // If spanPolicyValid is true and spanPolicyEndKey is empty, then spanPolicy
    3348            2 :         // applies for the rest of the keyspace.
    3349            2 :         var spanPolicyEndKey []byte
    3350            2 : 
    3351            2 :         for runner.MoreDataToWrite() {
    3352            2 :                 if c.cancel.Load() {
    3353            2 :                         return runner.Finish().WithError(ErrCancelledCompaction)
    3354            2 :                 }
    3355              :                 // Create a new table.
    3356            2 :                 firstKey := runner.FirstKey()
    3357            2 :                 if !spanPolicyValid || (len(spanPolicyEndKey) > 0 && d.cmp(firstKey, spanPolicyEndKey) >= 0) {
    3358            2 :                         var err error
    3359            2 :                         spanPolicy, spanPolicyEndKey, err = d.opts.Experimental.SpanPolicyFunc(firstKey)
    3360            2 :                         if err != nil {
    3361            0 :                                 return runner.Finish().WithError(err)
    3362            0 :                         }
    3363            2 :                         spanPolicyValid = true
    3364              :                 }
    3365              : 
    3366            2 :                 writerOpts := d.opts.MakeWriterOptions(c.outputLevel.level, tableFormat)
    3367            2 :                 if spanPolicy.DisableValueSeparationBySuffix {
    3368            1 :                         writerOpts.DisableValueBlocks = true
    3369            1 :                 }
    3370            2 :                 if spanPolicy.PreferFastCompression && writerOpts.Compression != block.NoCompression {
    3371            0 :                         writerOpts.Compression = block.FastestCompression
    3372            0 :                 }
    3373            2 :                 vSep := valueSeparation
    3374            2 :                 if spanPolicy.ValueStoragePolicy == ValueStorageLowReadLatency {
    3375            1 :                         vSep = compact.NeverSeparateValues{}
    3376            1 :                 }
    3377            2 :                 objMeta, tw, err := d.newCompactionOutputTable(jobID, c, writerOpts)
    3378            2 :                 if err != nil {
    3379            1 :                         return runner.Finish().WithError(err)
    3380            1 :                 }
    3381            2 :                 runner.WriteTable(objMeta, tw, spanPolicyEndKey, vSep)
    3382              :         }
    3383            2 :         result = runner.Finish()
    3384            2 :         if result.Err == nil {
    3385            2 :                 result.Err = d.objProvider.Sync()
    3386            2 :         }
    3387            2 :         return result
    3388              : }
    3389              : 
    3390              : // makeVersionEdit creates the version edit for a compaction, based on the
    3391              : // tables in compact.Result.
    3392            2 : func (c *tableCompaction) makeVersionEdit(result compact.Result) (*manifest.VersionEdit, error) {
    3393            2 :         ve := &manifest.VersionEdit{
    3394            2 :                 DeletedTables: map[manifest.DeletedTableEntry]*manifest.TableMetadata{},
    3395            2 :         }
    3396            2 :         for _, cl := range c.inputs {
    3397            2 :                 for f := range cl.files.All() {
    3398            2 :                         ve.DeletedTables[manifest.DeletedTableEntry{
    3399            2 :                                 Level:   cl.level,
    3400            2 :                                 FileNum: f.TableNum,
    3401            2 :                         }] = f
    3402            2 :                 }
    3403              :         }
    3404              :         // Add any newly constructed blob files to the version edit.
    3405            2 :         ve.NewBlobFiles = make([]manifest.BlobFileMetadata, len(result.Blobs))
    3406            2 :         for i := range result.Blobs {
    3407            2 :                 ve.NewBlobFiles[i] = manifest.BlobFileMetadata{
    3408            2 :                         FileID:   base.BlobFileID(result.Blobs[i].Metadata.FileNum),
    3409            2 :                         Physical: result.Blobs[i].Metadata,
    3410            2 :                 }
    3411            2 :         }
    3412              : 
    3413            2 :         startLevelBytes := c.startLevel.files.TableSizeSum()
    3414            2 : 
    3415            2 :         outputMetrics := c.metrics.perLevel.level(c.outputLevel.level)
    3416            2 :         outputMetrics.TableBytesIn = startLevelBytes
    3417            2 :         // TODO(jackson):  This BytesRead value does not include any blob files
    3418            2 :         // written. It either should, or we should add a separate metric.
    3419            2 :         outputMetrics.TableBytesRead = c.outputLevel.files.TableSizeSum()
    3420            2 :         outputMetrics.BlobBytesCompacted = result.Stats.CumulativeBlobFileSize
    3421            2 :         if c.flush.flushables != nil {
    3422            2 :                 outputMetrics.BlobBytesFlushed = result.Stats.CumulativeBlobFileSize
    3423            2 :         }
    3424            2 :         if len(c.extraLevels) > 0 {
    3425            2 :                 outputMetrics.TableBytesIn += c.extraLevels[0].files.TableSizeSum()
    3426            2 :         }
    3427            2 :         outputMetrics.TableBytesRead += outputMetrics.TableBytesIn
    3428            2 : 
    3429            2 :         if len(c.flush.flushables) == 0 {
    3430            2 :                 c.metrics.perLevel.level(c.startLevel.level)
    3431            2 :         }
    3432            2 :         if len(c.extraLevels) > 0 {
    3433            2 :                 c.metrics.perLevel.level(c.extraLevels[0].level)
    3434            2 :                 outputMetrics.MultiLevel.TableBytesInTop = startLevelBytes
    3435            2 :                 outputMetrics.MultiLevel.TableBytesIn = outputMetrics.TableBytesIn
    3436            2 :                 outputMetrics.MultiLevel.TableBytesRead = outputMetrics.TableBytesRead
    3437            2 :         }
    3438              : 
    3439            2 :         inputLargestSeqNumAbsolute := c.inputLargestSeqNumAbsolute()
    3440            2 :         ve.NewTables = make([]manifest.NewTableEntry, len(result.Tables))
    3441            2 :         for i := range result.Tables {
    3442            2 :                 t := &result.Tables[i]
    3443            2 : 
    3444            2 :                 if t.WriterMeta.Properties.NumValuesInBlobFiles > 0 {
    3445            2 :                         if len(t.BlobReferences) == 0 {
    3446            0 :                                 return nil, base.AssertionFailedf("num values in blob files %d but no blob references",
    3447            0 :                                         t.WriterMeta.Properties.NumValuesInBlobFiles)
    3448            0 :                         }
    3449              :                 }
    3450              : 
    3451            2 :                 fileMeta := &manifest.TableMetadata{
    3452            2 :                         TableNum:           base.PhysicalTableFileNum(t.ObjMeta.DiskFileNum),
    3453            2 :                         CreationTime:       t.CreationTime.Unix(),
    3454            2 :                         Size:               t.WriterMeta.Size,
    3455            2 :                         SmallestSeqNum:     t.WriterMeta.SmallestSeqNum,
    3456            2 :                         LargestSeqNum:      t.WriterMeta.LargestSeqNum,
    3457            2 :                         BlobReferences:     t.BlobReferences,
    3458            2 :                         BlobReferenceDepth: t.BlobReferenceDepth,
    3459            2 :                 }
    3460            2 :                 if c.flush.flushables == nil {
    3461            2 :                         // Set the file's LargestSeqNumAbsolute to be the maximum value of any
    3462            2 :                         // of the compaction's input sstables.
    3463            2 :                         // TODO(jackson): This could be narrowed to be the maximum of input
    3464            2 :                         // sstables that overlap the output sstable's key range.
    3465            2 :                         fileMeta.LargestSeqNumAbsolute = inputLargestSeqNumAbsolute
    3466            2 :                 } else {
    3467            2 :                         fileMeta.LargestSeqNumAbsolute = t.WriterMeta.LargestSeqNum
    3468            2 :                 }
    3469            2 :                 fileMeta.InitPhysicalBacking()
    3470            2 : 
    3471            2 :                 // If the file didn't contain any range deletions, we can fill its
    3472            2 :                 // table stats now, avoiding unnecessarily loading the table later.
    3473            2 :                 maybeSetStatsFromProperties(
    3474            2 :                         fileMeta.PhysicalMeta(), &t.WriterMeta.Properties.CommonProperties, c.logger,
    3475            2 :                 )
    3476            2 : 
    3477            2 :                 if t.WriterMeta.HasPointKeys {
    3478            2 :                         fileMeta.ExtendPointKeyBounds(c.comparer.Compare,
    3479            2 :                                 t.WriterMeta.SmallestPoint,
    3480            2 :                                 t.WriterMeta.LargestPoint)
    3481            2 :                 }
    3482            2 :                 if t.WriterMeta.HasRangeDelKeys {
    3483            2 :                         fileMeta.ExtendPointKeyBounds(c.comparer.Compare,
    3484            2 :                                 t.WriterMeta.SmallestRangeDel,
    3485            2 :                                 t.WriterMeta.LargestRangeDel)
    3486            2 :                 }
    3487            2 :                 if t.WriterMeta.HasRangeKeys {
    3488            2 :                         fileMeta.ExtendRangeKeyBounds(c.comparer.Compare,
    3489            2 :                                 t.WriterMeta.SmallestRangeKey,
    3490            2 :                                 t.WriterMeta.LargestRangeKey)
    3491            2 :                 }
    3492              : 
    3493            2 :                 ve.NewTables[i] = manifest.NewTableEntry{
    3494            2 :                         Level: c.outputLevel.level,
    3495            2 :                         Meta:  fileMeta,
    3496            2 :                 }
    3497            2 : 
    3498            2 :                 // Update metrics.
    3499            2 :                 if c.flush.flushables == nil {
    3500            2 :                         outputMetrics.TablesCompacted++
    3501            2 :                         outputMetrics.TableBytesCompacted += fileMeta.Size
    3502            2 :                 } else {
    3503            2 :                         outputMetrics.TablesFlushed++
    3504            2 :                         outputMetrics.TableBytesFlushed += fileMeta.Size
    3505            2 :                 }
    3506            2 :                 outputMetrics.EstimatedReferencesSize += fileMeta.EstimatedReferenceSize()
    3507            2 :                 outputMetrics.BlobBytesReadEstimate += fileMeta.EstimatedReferenceSize()
    3508            2 :                 outputMetrics.TablesSize += int64(fileMeta.Size)
    3509            2 :                 outputMetrics.TablesCount++
    3510            2 :                 outputMetrics.Additional.BytesWrittenDataBlocks += t.WriterMeta.Properties.DataSize
    3511            2 :                 outputMetrics.Additional.BytesWrittenValueBlocks += t.WriterMeta.Properties.ValueBlocksSize
    3512              :         }
    3513              : 
    3514              :         // Sanity check that the tables are ordered and don't overlap.
    3515            2 :         for i := 1; i < len(ve.NewTables); i++ {
    3516            2 :                 if ve.NewTables[i-1].Meta.Largest().IsUpperBoundFor(c.comparer.Compare, ve.NewTables[i].Meta.Smallest().UserKey) {
    3517            0 :                         return nil, base.AssertionFailedf("pebble: compaction output tables overlap: %s and %s",
    3518            0 :                                 ve.NewTables[i-1].Meta.DebugString(c.comparer.FormatKey, true),
    3519            0 :                                 ve.NewTables[i].Meta.DebugString(c.comparer.FormatKey, true),
    3520            0 :                         )
    3521            0 :                 }
    3522              :         }
    3523              : 
    3524            2 :         return ve, nil
    3525              : }
    3526              : 
    3527              : // newCompactionOutputTable creates an object for a new table produced by a
    3528              : // compaction or flush.
    3529              : func (d *DB) newCompactionOutputTable(
    3530              :         jobID JobID, c *tableCompaction, writerOpts sstable.WriterOptions,
    3531            2 : ) (objstorage.ObjectMetadata, sstable.RawWriter, error) {
    3532            2 :         writable, objMeta, err := d.newCompactionOutputObj(c, base.FileTypeTable)
    3533            2 :         if err != nil {
    3534            1 :                 return objstorage.ObjectMetadata{}, nil, err
    3535            1 :         }
    3536            2 :         d.opts.EventListener.TableCreated(TableCreateInfo{
    3537            2 :                 JobID:   int(jobID),
    3538            2 :                 Reason:  c.kind.compactingOrFlushing(),
    3539            2 :                 Path:    d.objProvider.Path(objMeta),
    3540            2 :                 FileNum: objMeta.DiskFileNum,
    3541            2 :         })
    3542            2 :         writerOpts.SetInternal(sstableinternal.WriterOptions{
    3543            2 :                 CacheOpts: sstableinternal.CacheOptions{
    3544            2 :                         CacheHandle: d.cacheHandle,
    3545            2 :                         FileNum:     objMeta.DiskFileNum,
    3546            2 :                 },
    3547            2 :         })
    3548            2 :         tw := sstable.NewRawWriterWithCPUMeasurer(writable, writerOpts, c.grantHandle)
    3549            2 :         return objMeta, tw, nil
    3550              : }
    3551              : 
    3552              : // newCompactionOutputBlob creates an object for a new blob produced by a
    3553              : // compaction or flush.
    3554              : func (d *DB) newCompactionOutputBlob(
    3555              :         jobID JobID, c *tableCompaction,
    3556            2 : ) (objstorage.Writable, objstorage.ObjectMetadata, error) {
    3557            2 :         writable, objMeta, err := d.newCompactionOutputObj(c, base.FileTypeBlob)
    3558            2 :         if err != nil {
    3559            0 :                 return nil, objstorage.ObjectMetadata{}, err
    3560            0 :         }
    3561            2 :         d.opts.EventListener.BlobFileCreated(BlobFileCreateInfo{
    3562            2 :                 JobID:   int(jobID),
    3563            2 :                 Reason:  c.kind.compactingOrFlushing(),
    3564            2 :                 Path:    d.objProvider.Path(objMeta),
    3565            2 :                 FileNum: objMeta.DiskFileNum,
    3566            2 :         })
    3567            2 :         return writable, objMeta, nil
    3568              : }
    3569              : 
    3570              : // newCompactionOutputObj creates an object produced by a compaction or flush.
    3571              : func (d *DB) newCompactionOutputObj(
    3572              :         c *tableCompaction, typ base.FileType,
    3573            2 : ) (objstorage.Writable, objstorage.ObjectMetadata, error) {
    3574            2 :         diskFileNum := d.mu.versions.getNextDiskFileNum()
    3575            2 :         ctx := context.TODO()
    3576            2 : 
    3577            2 :         if objiotracing.Enabled {
    3578            0 :                 ctx = objiotracing.WithLevel(ctx, c.outputLevel.level)
    3579            0 :                 if c.kind == compactionKindFlush {
    3580            0 :                         ctx = objiotracing.WithReason(ctx, objiotracing.ForFlush)
    3581            0 :                 } else {
    3582            0 :                         ctx = objiotracing.WithReason(ctx, objiotracing.ForCompaction)
    3583            0 :                 }
    3584              :         }
    3585              : 
    3586            2 :         writable, objMeta, err := d.objProvider.Create(ctx, typ, diskFileNum, c.objCreateOpts)
    3587            2 :         if err != nil {
    3588            1 :                 return nil, objstorage.ObjectMetadata{}, err
    3589            1 :         }
    3590              : 
    3591            2 :         if c.kind != compactionKindFlush {
    3592            2 :                 writable = &compactionWritable{
    3593            2 :                         Writable: writable,
    3594            2 :                         versions: d.mu.versions,
    3595            2 :                         written:  &c.metrics.bytesWritten,
    3596            2 :                 }
    3597            2 :         }
    3598            2 :         return writable, objMeta, nil
    3599              : }
    3600              : 
    3601              : // validateVersionEdit validates that start and end keys across new and deleted
    3602              : // files in a versionEdit pass the given validation function.
    3603              : func validateVersionEdit(
    3604              :         ve *manifest.VersionEdit, vk base.ValidateKey, format base.FormatKey, logger Logger,
    3605            2 : ) {
    3606            2 :         validateKey := func(f *manifest.TableMetadata, key []byte) {
    3607            2 :                 if err := vk.Validate(key); err != nil {
    3608            1 :                         logger.Fatalf("pebble: version edit validation failed (key=%s file=%s): %v", format(key), f, err)
    3609            1 :                 }
    3610              :         }
    3611              : 
    3612              :         // Validate both new and deleted files.
    3613            2 :         for _, f := range ve.NewTables {
    3614            2 :                 validateKey(f.Meta, f.Meta.Smallest().UserKey)
    3615            2 :                 validateKey(f.Meta, f.Meta.Largest().UserKey)
    3616            2 :         }
    3617            2 :         for _, m := range ve.DeletedTables {
    3618            2 :                 validateKey(m, m.Smallest().UserKey)
    3619            2 :                 validateKey(m, m.Largest().UserKey)
    3620            2 :         }
    3621              : }
    3622              : 
    3623            2 : func getDiskWriteCategoryForCompaction(opts *Options, kind compactionKind) vfs.DiskWriteCategory {
    3624            2 :         if opts.EnableSQLRowSpillMetrics {
    3625            0 :                 // In the scenario that the Pebble engine is used for SQL row spills the
    3626            0 :                 // data written to the memtable will correspond to spills to disk and
    3627            0 :                 // should be categorized as such.
    3628            0 :                 return "sql-row-spill"
    3629            2 :         } else if kind == compactionKindFlush {
    3630            2 :                 return "pebble-memtable-flush"
    3631            2 :         } else if kind == compactionKindBlobFileRewrite {
    3632            1 :                 return "pebble-blob-file-rewrite"
    3633            2 :         } else {
    3634            2 :                 return "pebble-compaction"
    3635            2 :         }
    3636              : }
        

Generated by: LCOV version 2.0-1