LCOV - code coverage report
Current view: top level - pebble - ingest.go (source / functions) Coverage Total Hit
Test: 2025-06-20 08:19Z 54bb06f3 - tests + meta.lcov Lines: 87.4 % 1541 1347
Test Date: 2025-06-20 08:20:18 Functions: - 0 0

            Line data    Source code
       1              : // Copyright 2018 The LevelDB-Go and Pebble Authors. All rights reserved. Use
       2              : // of this source code is governed by a BSD-style license that can be found in
       3              : // the LICENSE file.
       4              : 
       5              : package pebble
       6              : 
       7              : import (
       8              :         "context"
       9              :         "fmt"
      10              :         "slices"
      11              :         "sort"
      12              :         "time"
      13              : 
      14              :         "github.com/cockroachdb/errors"
      15              :         "github.com/cockroachdb/pebble/internal/base"
      16              :         "github.com/cockroachdb/pebble/internal/cache"
      17              :         "github.com/cockroachdb/pebble/internal/invariants"
      18              :         "github.com/cockroachdb/pebble/internal/keyspan"
      19              :         "github.com/cockroachdb/pebble/internal/manifest"
      20              :         "github.com/cockroachdb/pebble/internal/overlap"
      21              :         "github.com/cockroachdb/pebble/internal/sstableinternal"
      22              :         "github.com/cockroachdb/pebble/objstorage"
      23              :         "github.com/cockroachdb/pebble/objstorage/remote"
      24              :         "github.com/cockroachdb/pebble/sstable"
      25              :         "github.com/cockroachdb/pebble/sstable/block"
      26              : )
      27              : 
      28            2 : func sstableKeyCompare(userCmp Compare, a, b InternalKey) int {
      29            2 :         c := userCmp(a.UserKey, b.UserKey)
      30            2 :         if c != 0 {
      31            2 :                 return c
      32            2 :         }
      33            2 :         if a.IsExclusiveSentinel() {
      34            2 :                 if !b.IsExclusiveSentinel() {
      35            2 :                         return -1
      36            2 :                 }
      37            2 :         } else if b.IsExclusiveSentinel() {
      38            2 :                 return +1
      39            2 :         }
      40            2 :         return 0
      41              : }
      42              : 
      43            2 : func ingestValidateKey(opts *Options, key *InternalKey) error {
      44            2 :         if key.Kind() == InternalKeyKindInvalid {
      45            1 :                 return base.CorruptionErrorf("pebble: external sstable has corrupted key: %s",
      46            1 :                         key.Pretty(opts.Comparer.FormatKey))
      47            1 :         }
      48            2 :         if key.SeqNum() != 0 {
      49            1 :                 return base.CorruptionErrorf("pebble: external sstable has non-zero seqnum: %s",
      50            1 :                         key.Pretty(opts.Comparer.FormatKey))
      51            1 :         }
      52            2 :         if err := opts.Comparer.ValidateKey.Validate(key.UserKey); err != nil {
      53            0 :                 return base.CorruptionErrorf("pebble: external sstable has corrupted key: %s, %w",
      54            0 :                         key.Pretty(opts.Comparer.FormatKey), err)
      55            0 :         }
      56            2 :         return nil
      57              : }
      58              : 
      59              : // ingestSynthesizeShared constructs a fileMetadata for one shared sstable owned
      60              : // or shared by another node.
      61              : func ingestSynthesizeShared(
      62              :         opts *Options, sm SharedSSTMeta, tableNum base.TableNum,
      63            2 : ) (*manifest.TableMetadata, error) {
      64            2 :         if sm.Size == 0 {
      65            0 :                 // Disallow 0 file sizes
      66            0 :                 return nil, errors.New("pebble: cannot ingest shared file with size 0")
      67            0 :         }
      68              :         // Don't load table stats. Doing a round trip to shared storage, one SST
      69              :         // at a time is not worth it as it slows down ingestion.
      70            2 :         meta := &manifest.TableMetadata{
      71            2 :                 TableNum:     tableNum,
      72            2 :                 CreationTime: time.Now().Unix(),
      73            2 :                 Virtual:      true,
      74            2 :                 Size:         sm.Size,
      75            2 :         }
      76            2 :         if sm.LargestPointKey.Valid() && sm.LargestPointKey.UserKey != nil {
      77            2 :                 // Initialize meta.{HasPointKeys,Smallest,Largest}, etc.
      78            2 :                 //
      79            2 :                 // NB: We create new internal keys and pass them into ExtendPointKeyBounds
      80            2 :                 // so that we can sub a zero sequence number into the bounds. We can set
      81            2 :                 // the sequence number to anything here; it'll be reset in ingestUpdateSeqNum
      82            2 :                 // anyway. However, we do need to use the same sequence number across all
      83            2 :                 // bound keys at this step so that we end up with bounds that are consistent
      84            2 :                 // across point/range keys.
      85            2 :                 //
      86            2 :                 // Because of the sequence number rewriting, we cannot use the Kind of
      87            2 :                 // sm.SmallestPointKey. For example, the original SST might start with
      88            2 :                 // a.SET.2 and a.RANGEDEL.1 (with a.SET.2 being the smallest key); after
      89            2 :                 // rewriting the sequence numbers, these keys become a.SET.100 and
      90            2 :                 // a.RANGEDEL.100, with a.RANGEDEL.100 being the smallest key. To create a
      91            2 :                 // correct bound, we just use the maximum key kind (which sorts first).
      92            2 :                 // Similarly, we use the smallest key kind for the largest key.
      93            2 :                 smallestPointKey := base.MakeInternalKey(sm.SmallestPointKey.UserKey, 0, base.InternalKeyKindMaxForSSTable)
      94            2 :                 largestPointKey := base.MakeInternalKey(sm.LargestPointKey.UserKey, 0, 0)
      95            2 :                 if sm.LargestPointKey.IsExclusiveSentinel() {
      96            2 :                         largestPointKey = base.MakeRangeDeleteSentinelKey(sm.LargestPointKey.UserKey)
      97            2 :                 }
      98            2 :                 if opts.Comparer.Equal(smallestPointKey.UserKey, largestPointKey.UserKey) &&
      99            2 :                         smallestPointKey.Trailer < largestPointKey.Trailer {
     100            0 :                         // We get kinds from the sender, however we substitute our own sequence
     101            0 :                         // numbers. This can result in cases where an sstable [b#5,SET-b#4,DELSIZED]
     102            0 :                         // becomes [b#0,SET-b#0,DELSIZED] when we synthesize it here, but the
     103            0 :                         // kinds need to be reversed now because DelSized > Set.
     104            0 :                         smallestPointKey, largestPointKey = largestPointKey, smallestPointKey
     105            0 :                 }
     106            2 :                 meta.ExtendPointKeyBounds(opts.Comparer.Compare, smallestPointKey, largestPointKey)
     107              :         }
     108            2 :         if sm.LargestRangeKey.Valid() && sm.LargestRangeKey.UserKey != nil {
     109            2 :                 // Initialize meta.{HasRangeKeys,Smallest,Largest}, etc.
     110            2 :                 //
     111            2 :                 // See comment above on why we use a zero sequence number and these key
     112            2 :                 // kinds here.
     113            2 :                 smallestRangeKey := base.MakeInternalKey(sm.SmallestRangeKey.UserKey, 0, base.InternalKeyKindRangeKeyMax)
     114            2 :                 largestRangeKey := base.MakeExclusiveSentinelKey(base.InternalKeyKindRangeKeyMin, sm.LargestRangeKey.UserKey)
     115            2 :                 meta.ExtendRangeKeyBounds(opts.Comparer.Compare, smallestRangeKey, largestRangeKey)
     116            2 :         }
     117              : 
     118              :         // For simplicity, we use the same number for both the FileNum and the
     119              :         // DiskFileNum (even though this is a virtual sstable). Pass the underlying
     120              :         // TableBacking's size to the same size as the virtualized view of the sstable.
     121              :         // This ensures that we don't over-prioritize this sstable for compaction just
     122              :         // yet, as we do not have a clear sense of what parts of this sstable are
     123              :         // referenced by other nodes.
     124            2 :         meta.InitVirtualBacking(base.DiskFileNum(tableNum), sm.Size)
     125            2 : 
     126            2 :         if err := meta.Validate(opts.Comparer.Compare, opts.Comparer.FormatKey); err != nil {
     127            0 :                 return nil, err
     128            0 :         }
     129            2 :         return meta, nil
     130              : }
     131              : 
     132              : // ingestLoad1External loads the fileMetadata for one external sstable.
     133              : // Sequence number and target level calculation happens during prepare/apply.
     134              : func ingestLoad1External(
     135              :         opts *Options, e ExternalFile, tableNum base.TableNum,
     136            2 : ) (*manifest.TableMetadata, error) {
     137            2 :         if e.Size == 0 {
     138            0 :                 return nil, errors.New("pebble: cannot ingest external file with size 0")
     139            0 :         }
     140            2 :         if !e.HasRangeKey && !e.HasPointKey {
     141            0 :                 return nil, errors.New("pebble: cannot ingest external file with no point or range keys")
     142            0 :         }
     143              : 
     144            2 :         if opts.Comparer.Compare(e.StartKey, e.EndKey) > 0 {
     145            1 :                 return nil, errors.Newf("pebble: external file bounds [%q, %q) are invalid", e.StartKey, e.EndKey)
     146            1 :         }
     147            2 :         if opts.Comparer.Compare(e.StartKey, e.EndKey) == 0 && !e.EndKeyIsInclusive {
     148            0 :                 return nil, errors.Newf("pebble: external file bounds [%q, %q) are invalid", e.StartKey, e.EndKey)
     149            0 :         }
     150            2 :         if n := opts.Comparer.Split(e.StartKey); n != len(e.StartKey) {
     151            1 :                 return nil, errors.Newf("pebble: external file bounds start key %q has suffix", e.StartKey)
     152            1 :         }
     153            2 :         if n := opts.Comparer.Split(e.EndKey); n != len(e.EndKey) {
     154            1 :                 return nil, errors.Newf("pebble: external file bounds end key %q has suffix", e.EndKey)
     155            1 :         }
     156              : 
     157              :         // Don't load table stats. Doing a round trip to shared storage, one SST
     158              :         // at a time is not worth it as it slows down ingestion.
     159            2 :         meta := &manifest.TableMetadata{
     160            2 :                 TableNum:     tableNum,
     161            2 :                 CreationTime: time.Now().Unix(),
     162            2 :                 Size:         e.Size,
     163            2 :                 Virtual:      true,
     164            2 :         }
     165            2 :         // In the name of keeping this ingestion as fast as possible, we avoid *all*
     166            2 :         // existence checks and synthesize a table metadata with smallest/largest
     167            2 :         // keys that overlap whatever the passed-in span was.
     168            2 :         smallestCopy := slices.Clone(e.StartKey)
     169            2 :         largestCopy := slices.Clone(e.EndKey)
     170            2 :         if e.HasPointKey {
     171            2 :                 // Sequence numbers are updated later by
     172            2 :                 // ingestUpdateSeqNum, applying a squence number that
     173            2 :                 // is applied to all keys in the sstable.
     174            2 :                 if e.EndKeyIsInclusive {
     175            1 :                         meta.ExtendPointKeyBounds(
     176            1 :                                 opts.Comparer.Compare,
     177            1 :                                 base.MakeInternalKey(smallestCopy, 0, base.InternalKeyKindMaxForSSTable),
     178            1 :                                 base.MakeInternalKey(largestCopy, 0, 0))
     179            2 :                 } else {
     180            2 :                         meta.ExtendPointKeyBounds(
     181            2 :                                 opts.Comparer.Compare,
     182            2 :                                 base.MakeInternalKey(smallestCopy, 0, base.InternalKeyKindMaxForSSTable),
     183            2 :                                 base.MakeRangeDeleteSentinelKey(largestCopy))
     184            2 :                 }
     185              :         }
     186            2 :         if e.HasRangeKey {
     187            2 :                 meta.ExtendRangeKeyBounds(
     188            2 :                         opts.Comparer.Compare,
     189            2 :                         base.MakeInternalKey(smallestCopy, 0, InternalKeyKindRangeKeyMax),
     190            2 :                         base.MakeExclusiveSentinelKey(InternalKeyKindRangeKeyMin, largestCopy),
     191            2 :                 )
     192            2 :         }
     193              : 
     194            2 :         meta.SyntheticPrefixAndSuffix = sstable.MakeSyntheticPrefixAndSuffix(e.SyntheticPrefix, e.SyntheticSuffix)
     195            2 : 
     196            2 :         return meta, nil
     197              : }
     198              : 
     199              : type rangeKeyIngestValidator struct {
     200              :         // lastRangeKey is the last range key seen in the previous file.
     201              :         lastRangeKey keyspan.Span
     202              :         // comparer, if unset, disables range key validation.
     203              :         comparer *base.Comparer
     204              : }
     205              : 
     206            2 : func disableRangeKeyChecks() rangeKeyIngestValidator {
     207            2 :         return rangeKeyIngestValidator{}
     208            2 : }
     209              : 
     210              : func validateSuffixedBoundaries(
     211              :         cmp *base.Comparer, lastRangeKey keyspan.Span,
     212            2 : ) rangeKeyIngestValidator {
     213            2 :         return rangeKeyIngestValidator{
     214            2 :                 lastRangeKey: lastRangeKey,
     215            2 :                 comparer:     cmp,
     216            2 :         }
     217            2 : }
     218              : 
     219              : // Validate valides if the stored state of this rangeKeyIngestValidator allows for
     220              : // a file with the given nextFileSmallestKey to be ingested, such that the stored
     221              : // last file's largest range key defragments cleanly with the next file's smallest
     222              : // key if it was suffixed. If a value of nil is passed in for nextFileSmallestKey,
     223              : // that denotes the next file does not have a range key or there is no next file.
     224            2 : func (r *rangeKeyIngestValidator) Validate(nextFileSmallestKey *keyspan.Span) error {
     225            2 :         if r.comparer == nil {
     226            2 :                 return nil
     227            2 :         }
     228            2 :         if r.lastRangeKey.Valid() {
     229            2 :                 if r.comparer.Split.HasSuffix(r.lastRangeKey.End) {
     230            1 :                         if nextFileSmallestKey == nil || !r.comparer.Equal(r.lastRangeKey.End, nextFileSmallestKey.Start) {
     231            1 :                                 // The last range key has a suffix, and it doesn't defragment cleanly with this range key.
     232            1 :                                 return errors.AssertionFailedf("pebble: ingest sstable has suffixed largest range key that does not match the start key of the next sstable: %s",
     233            1 :                                         r.comparer.FormatKey(r.lastRangeKey.End))
     234            1 :                         } else if !keyspan.DefragmentInternal.ShouldDefragment(r.comparer.CompareRangeSuffixes, &r.lastRangeKey, nextFileSmallestKey) {
     235            0 :                                 // The last range key has a suffix, and it doesn't defragment cleanly with this range key.
     236            0 :                                 return errors.AssertionFailedf("pebble: ingest sstable has suffixed range key that won't defragment with next sstable: %s",
     237            0 :                                         r.comparer.FormatKey(r.lastRangeKey.End))
     238            0 :                         }
     239              :                 }
     240            2 :         } else if nextFileSmallestKey != nil && r.comparer.Split.HasSuffix(nextFileSmallestKey.Start) {
     241            0 :                 return errors.Newf("pebble: ingest sstable has suffixed range key start that won't defragment: %s",
     242            0 :                         r.comparer.FormatKey(nextFileSmallestKey.Start))
     243            0 :         }
     244            2 :         return nil
     245              : }
     246              : 
     247              : // ingestLoad1 creates the TableMetadata for one file. This file will be owned
     248              : // by this store.
     249              : //
     250              : // prevLastRangeKey is the last range key from the previous file. It is used to
     251              : // ensure that the range keys defragment cleanly across files. These checks
     252              : // are disabled if disableRangeKeyChecks is true.
     253              : func ingestLoad1(
     254              :         ctx context.Context,
     255              :         opts *Options,
     256              :         fmv FormatMajorVersion,
     257              :         readable objstorage.Readable,
     258              :         cacheHandle *cache.Handle,
     259              :         tableNum base.TableNum,
     260              :         rangeKeyValidator rangeKeyIngestValidator,
     261            2 : ) (meta *manifest.TableMetadata, lastRangeKey keyspan.Span, err error) {
     262            2 :         o := opts.MakeReaderOptions()
     263            2 :         o.CacheOpts = sstableinternal.CacheOptions{
     264            2 :                 CacheHandle: cacheHandle,
     265            2 :                 FileNum:     base.PhysicalTableDiskFileNum(tableNum),
     266            2 :         }
     267            2 :         r, err := sstable.NewReader(ctx, readable, o)
     268            2 :         if err != nil {
     269            1 :                 return nil, keyspan.Span{}, errors.CombineErrors(err, readable.Close())
     270            1 :         }
     271            2 :         defer func() { _ = r.Close() }()
     272              : 
     273              :         // Avoid ingesting tables with format versions this DB doesn't support.
     274            2 :         tf, err := r.TableFormat()
     275            2 :         if err != nil {
     276            0 :                 return nil, keyspan.Span{}, err
     277            0 :         }
     278            2 :         if tf < fmv.MinTableFormat() || tf > fmv.MaxTableFormat() {
     279            1 :                 return nil, keyspan.Span{}, errors.Newf(
     280            1 :                         "pebble: table format %s is not within range supported at DB format major version %d, (%s,%s)",
     281            1 :                         tf, fmv, fmv.MinTableFormat(), fmv.MaxTableFormat(),
     282            1 :                 )
     283            1 :         }
     284              : 
     285            2 :         if r.Attributes.Has(sstable.AttributeBlobValues) {
     286            1 :                 return nil, keyspan.Span{}, errors.Newf(
     287            1 :                         "pebble: ingesting tables with blob references is not supported")
     288            1 :         }
     289              : 
     290            2 :         props, err := r.ReadPropertiesBlock(ctx, nil /* buffer pool */)
     291            2 :         if err != nil {
     292            1 :                 return nil, keyspan.Span{}, err
     293            1 :         }
     294              : 
     295              :         // If this is a columnar block, read key schema name from properties block.
     296            2 :         if tf.BlockColumnar() {
     297            2 :                 if _, ok := opts.KeySchemas[props.KeySchemaName]; !ok {
     298            0 :                         return nil, keyspan.Span{}, errors.Newf(
     299            0 :                                 "pebble: table uses key schema %q unknown to the database",
     300            0 :                                 props.KeySchemaName)
     301            0 :                 }
     302              :         }
     303              : 
     304            2 :         meta = &manifest.TableMetadata{}
     305            2 :         meta.TableNum = tableNum
     306            2 :         meta.Size = max(uint64(readable.Size()), 1)
     307            2 :         meta.CreationTime = time.Now().Unix()
     308            2 :         meta.InitPhysicalBacking()
     309            2 : 
     310            2 :         // Avoid loading into the file cache for collecting stats if we
     311            2 :         // don't need to. If there are no range deletions, we have all the
     312            2 :         // information to compute the stats here.
     313            2 :         //
     314            2 :         // This is helpful in tests for avoiding awkwardness around deletion of
     315            2 :         // ingested files from MemFS. MemFS implements the Windows semantics of
     316            2 :         // disallowing removal of an open file. Under MemFS, if we don't populate
     317            2 :         // meta.Stats here, the file will be loaded into the file cache for
     318            2 :         // calculating stats before we can remove the original link.
     319            2 :         maybeSetStatsFromProperties(meta.PhysicalMeta(), &props.CommonProperties, opts.Logger)
     320            2 : 
     321            2 :         {
     322            2 :                 iter, err := r.NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
     323            2 :                 if err != nil {
     324            1 :                         return nil, keyspan.Span{}, err
     325            1 :                 }
     326            2 :                 defer func() { _ = iter.Close() }()
     327            2 :                 var smallest InternalKey
     328            2 :                 if kv := iter.First(); kv != nil {
     329            2 :                         if err := ingestValidateKey(opts, &kv.K); err != nil {
     330            1 :                                 return nil, keyspan.Span{}, err
     331            1 :                         }
     332            2 :                         smallest = kv.K.Clone()
     333              :                 }
     334            2 :                 if err := iter.Error(); err != nil {
     335            1 :                         return nil, keyspan.Span{}, err
     336            1 :                 }
     337            2 :                 if kv := iter.Last(); kv != nil {
     338            2 :                         if err := ingestValidateKey(opts, &kv.K); err != nil {
     339            0 :                                 return nil, keyspan.Span{}, err
     340            0 :                         }
     341            2 :                         meta.ExtendPointKeyBounds(opts.Comparer.Compare, smallest, kv.K.Clone())
     342              :                 }
     343            2 :                 if err := iter.Error(); err != nil {
     344            1 :                         return nil, keyspan.Span{}, err
     345            1 :                 }
     346              :         }
     347              : 
     348            2 :         iter, err := r.NewRawRangeDelIter(ctx, sstable.NoFragmentTransforms, sstable.NoReadEnv)
     349            2 :         if err != nil {
     350            0 :                 return nil, keyspan.Span{}, err
     351            0 :         }
     352            2 :         if iter != nil {
     353            2 :                 defer iter.Close()
     354            2 :                 var smallest InternalKey
     355            2 :                 if s, err := iter.First(); err != nil {
     356            0 :                         return nil, keyspan.Span{}, err
     357            2 :                 } else if s != nil {
     358            2 :                         key := s.SmallestKey()
     359            2 :                         if err := ingestValidateKey(opts, &key); err != nil {
     360            0 :                                 return nil, keyspan.Span{}, err
     361            0 :                         }
     362            2 :                         smallest = key.Clone()
     363              :                 }
     364            2 :                 if s, err := iter.Last(); err != nil {
     365            0 :                         return nil, keyspan.Span{}, err
     366            2 :                 } else if s != nil {
     367            2 :                         k := s.SmallestKey()
     368            2 :                         if err := ingestValidateKey(opts, &k); err != nil {
     369            0 :                                 return nil, keyspan.Span{}, err
     370            0 :                         }
     371            2 :                         largest := s.LargestKey().Clone()
     372            2 :                         meta.ExtendPointKeyBounds(opts.Comparer.Compare, smallest, largest)
     373              :                 }
     374              :         }
     375              : 
     376              :         // Update the range-key bounds for the table.
     377            2 :         {
     378            2 :                 iter, err := r.NewRawRangeKeyIter(ctx, sstable.NoFragmentTransforms, sstable.NoReadEnv)
     379            2 :                 if err != nil {
     380            0 :                         return nil, keyspan.Span{}, err
     381            0 :                 }
     382            2 :                 if iter != nil {
     383            2 :                         defer iter.Close()
     384            2 :                         var smallest InternalKey
     385            2 :                         if s, err := iter.First(); err != nil {
     386            0 :                                 return nil, keyspan.Span{}, err
     387            2 :                         } else if s != nil {
     388            2 :                                 key := s.SmallestKey()
     389            2 :                                 if err := ingestValidateKey(opts, &key); err != nil {
     390            0 :                                         return nil, keyspan.Span{}, err
     391            0 :                                 }
     392            2 :                                 smallest = key.Clone()
     393            2 :                                 // Range keys need some additional validation as we need to ensure they
     394            2 :                                 // defragment cleanly with the lastRangeKey from the previous file.
     395            2 :                                 if err := rangeKeyValidator.Validate(s); err != nil {
     396            0 :                                         return nil, keyspan.Span{}, err
     397            0 :                                 }
     398              :                         }
     399            2 :                         lastRangeKey = keyspan.Span{}
     400            2 :                         if s, err := iter.Last(); err != nil {
     401            0 :                                 return nil, keyspan.Span{}, err
     402            2 :                         } else if s != nil {
     403            2 :                                 k := s.SmallestKey()
     404            2 :                                 if err := ingestValidateKey(opts, &k); err != nil {
     405            0 :                                         return nil, keyspan.Span{}, err
     406            0 :                                 }
     407              :                                 // As range keys are fragmented, the end key of the last range key in
     408              :                                 // the table provides the upper bound for the table.
     409            2 :                                 largest := s.LargestKey().Clone()
     410            2 :                                 meta.ExtendRangeKeyBounds(opts.Comparer.Compare, smallest, largest)
     411            2 :                                 lastRangeKey = s.Clone()
     412            0 :                         } else {
     413            0 :                                 // s == nil.
     414            0 :                                 if err := rangeKeyValidator.Validate(nil /* nextFileSmallestKey */); err != nil {
     415            0 :                                         return nil, keyspan.Span{}, err
     416            0 :                                 }
     417              :                         }
     418            2 :                 } else {
     419            2 :                         if err := rangeKeyValidator.Validate(nil /* nextFileSmallestKey */); err != nil {
     420            0 :                                 return nil, keyspan.Span{}, err
     421            0 :                         }
     422            2 :                         lastRangeKey = keyspan.Span{}
     423              :                 }
     424              :         }
     425              : 
     426            2 :         if !meta.HasPointKeys && !meta.HasRangeKeys {
     427            2 :                 return nil, keyspan.Span{}, nil
     428            2 :         }
     429              : 
     430              :         // Sanity check that the various bounds on the file were set consistently.
     431            2 :         if err := meta.Validate(opts.Comparer.Compare, opts.Comparer.FormatKey); err != nil {
     432            0 :                 return nil, keyspan.Span{}, err
     433            0 :         }
     434              : 
     435            2 :         return meta, lastRangeKey, nil
     436              : }
     437              : 
     438              : type ingestLoadResult struct {
     439              :         local    []ingestLocalMeta
     440              :         shared   []ingestSharedMeta
     441              :         external []ingestExternalMeta
     442              : 
     443              :         externalFilesHaveLevel bool
     444              : }
     445              : 
     446              : type ingestLocalMeta struct {
     447              :         *manifest.TableMetadata
     448              :         path string
     449              : }
     450              : 
     451              : type ingestSharedMeta struct {
     452              :         *manifest.TableMetadata
     453              :         shared SharedSSTMeta
     454              : }
     455              : 
     456              : type ingestExternalMeta struct {
     457              :         *manifest.TableMetadata
     458              :         external ExternalFile
     459              :         // usedExistingBacking is true if the external file is reusing a backing
     460              :         // that existed before this ingestion. In this case, we called
     461              :         // VirtualBackings.Protect() on that backing; we will need to call
     462              :         // Unprotect() after the ingestion.
     463              :         usedExistingBacking bool
     464              : }
     465              : 
     466            2 : func (r *ingestLoadResult) fileCount() int {
     467            2 :         return len(r.local) + len(r.shared) + len(r.external)
     468            2 : }
     469              : 
     470              : func ingestLoad(
     471              :         ctx context.Context,
     472              :         opts *Options,
     473              :         fmv FormatMajorVersion,
     474              :         paths []string,
     475              :         shared []SharedSSTMeta,
     476              :         external []ExternalFile,
     477              :         cacheHandle *cache.Handle,
     478              :         pending []base.TableNum,
     479            2 : ) (ingestLoadResult, error) {
     480            2 :         localFileNums := pending[:len(paths)]
     481            2 :         sharedFileNums := pending[len(paths) : len(paths)+len(shared)]
     482            2 :         externalFileNums := pending[len(paths)+len(shared) : len(paths)+len(shared)+len(external)]
     483            2 : 
     484            2 :         var result ingestLoadResult
     485            2 :         result.local = make([]ingestLocalMeta, 0, len(paths))
     486            2 :         var lastRangeKey keyspan.Span
     487            2 :         // NB: we disable range key boundary assertions if we have shared or external files
     488            2 :         // present in this ingestion. This is because a suffixed range key in a local file
     489            2 :         // can possibly defragment with a suffixed range key in a shared or external file.
     490            2 :         // We also disable range key boundary assertions if we have CreateOnShared set to
     491            2 :         // true, as that means we could have suffixed RangeKeyDels or Unsets in the local
     492            2 :         // files that won't ever be surfaced, even if there are no shared or external files
     493            2 :         // in the ingestion.
     494            2 :         shouldDisableRangeKeyChecks := len(shared) > 0 || len(external) > 0 || opts.Experimental.CreateOnShared != remote.CreateOnSharedNone
     495            2 :         for i := range paths {
     496            2 :                 f, err := opts.FS.Open(paths[i])
     497            2 :                 if err != nil {
     498            1 :                         return ingestLoadResult{}, err
     499            1 :                 }
     500              : 
     501            2 :                 readable, err := sstable.NewSimpleReadable(f)
     502            2 :                 if err != nil {
     503            1 :                         return ingestLoadResult{}, err
     504            1 :                 }
     505            2 :                 var m *manifest.TableMetadata
     506            2 :                 rangeKeyValidator := disableRangeKeyChecks()
     507            2 :                 if !shouldDisableRangeKeyChecks {
     508            2 :                         rangeKeyValidator = validateSuffixedBoundaries(opts.Comparer, lastRangeKey)
     509            2 :                 }
     510            2 :                 m, lastRangeKey, err = ingestLoad1(ctx, opts, fmv, readable, cacheHandle, localFileNums[i], rangeKeyValidator)
     511            2 :                 if err != nil {
     512            1 :                         return ingestLoadResult{}, err
     513            1 :                 }
     514            2 :                 if m != nil {
     515            2 :                         result.local = append(result.local, ingestLocalMeta{
     516            2 :                                 TableMetadata: m,
     517            2 :                                 path:          paths[i],
     518            2 :                         })
     519            2 :                 }
     520              :         }
     521              : 
     522            2 :         if !shouldDisableRangeKeyChecks {
     523            2 :                 rangeKeyValidator := validateSuffixedBoundaries(opts.Comparer, lastRangeKey)
     524            2 :                 if err := rangeKeyValidator.Validate(nil /* nextFileSmallestKey */); err != nil {
     525            1 :                         return ingestLoadResult{}, err
     526            1 :                 }
     527              :         }
     528              : 
     529              :         // Sort the shared files according to level.
     530            2 :         sort.Sort(sharedByLevel(shared))
     531            2 : 
     532            2 :         result.shared = make([]ingestSharedMeta, 0, len(shared))
     533            2 :         for i := range shared {
     534            2 :                 m, err := ingestSynthesizeShared(opts, shared[i], sharedFileNums[i])
     535            2 :                 if err != nil {
     536            0 :                         return ingestLoadResult{}, err
     537            0 :                 }
     538            2 :                 if shared[i].Level < sharedLevelsStart {
     539            0 :                         return ingestLoadResult{}, errors.New("cannot ingest shared file in level below sharedLevelsStart")
     540            0 :                 }
     541            2 :                 result.shared = append(result.shared, ingestSharedMeta{
     542            2 :                         TableMetadata: m,
     543            2 :                         shared:        shared[i],
     544            2 :                 })
     545              :         }
     546            2 :         result.external = make([]ingestExternalMeta, 0, len(external))
     547            2 :         for i := range external {
     548            2 :                 m, err := ingestLoad1External(opts, external[i], externalFileNums[i])
     549            2 :                 if err != nil {
     550            1 :                         return ingestLoadResult{}, err
     551            1 :                 }
     552            2 :                 result.external = append(result.external, ingestExternalMeta{
     553            2 :                         TableMetadata: m,
     554            2 :                         external:      external[i],
     555            2 :                 })
     556            2 :                 if external[i].Level > 0 {
     557            1 :                         if i != 0 && !result.externalFilesHaveLevel {
     558            0 :                                 return ingestLoadResult{}, base.AssertionFailedf("pebble: external sstables must all have level set or unset")
     559            0 :                         }
     560            1 :                         result.externalFilesHaveLevel = true
     561            2 :                 } else if result.externalFilesHaveLevel {
     562            0 :                         return ingestLoadResult{}, base.AssertionFailedf("pebble: external sstables must all have level set or unset")
     563            0 :                 }
     564              :         }
     565            2 :         return result, nil
     566              : }
     567              : 
     568            2 : func ingestSortAndVerify(cmp Compare, lr ingestLoadResult, exciseSpan KeyRange) error {
     569            2 :         // Verify that all the shared files (i.e. files in sharedMeta)
     570            2 :         // fit within the exciseSpan.
     571            2 :         for _, f := range lr.shared {
     572            2 :                 if !exciseSpan.Contains(cmp, f.Smallest()) || !exciseSpan.Contains(cmp, f.Largest()) {
     573            0 :                         return errors.Newf("pebble: shared file outside of excise span, span [%s-%s), file = %s", exciseSpan.Start, exciseSpan.End, f.String())
     574            0 :                 }
     575              :         }
     576              : 
     577            2 :         if lr.externalFilesHaveLevel {
     578            1 :                 for _, f := range lr.external {
     579            1 :                         if !exciseSpan.Contains(cmp, f.Smallest()) || !exciseSpan.Contains(cmp, f.Largest()) {
     580            0 :                                 return base.AssertionFailedf("pebble: external file outside of excise span, span [%s-%s), file = %s", exciseSpan.Start, exciseSpan.End, f.String())
     581            0 :                         }
     582              :                 }
     583              :         }
     584              : 
     585            2 :         if len(lr.external) > 0 {
     586            2 :                 if len(lr.shared) > 0 {
     587            0 :                         // If external files are present alongside shared files,
     588            0 :                         // return an error.
     589            0 :                         return base.AssertionFailedf("pebble: external files cannot be ingested atomically alongside shared files")
     590            0 :                 }
     591              : 
     592              :                 // Sort according to the smallest key.
     593            2 :                 slices.SortFunc(lr.external, func(a, b ingestExternalMeta) int {
     594            2 :                         return cmp(a.Smallest().UserKey, b.Smallest().UserKey)
     595            2 :                 })
     596            2 :                 for i := 1; i < len(lr.external); i++ {
     597            2 :                         if sstableKeyCompare(cmp, lr.external[i-1].Largest(), lr.external[i].Smallest()) >= 0 {
     598            1 :                                 return errors.Newf("pebble: external sstables have overlapping ranges")
     599            1 :                         }
     600              :                 }
     601            2 :                 return nil
     602              :         }
     603            2 :         if len(lr.local) <= 1 {
     604            2 :                 return nil
     605            2 :         }
     606              : 
     607              :         // Sort according to the smallest key.
     608            2 :         slices.SortFunc(lr.local, func(a, b ingestLocalMeta) int {
     609            2 :                 return cmp(a.Smallest().UserKey, b.Smallest().UserKey)
     610            2 :         })
     611              : 
     612            2 :         for i := 1; i < len(lr.local); i++ {
     613            2 :                 if sstableKeyCompare(cmp, lr.local[i-1].Largest(), lr.local[i].Smallest()) >= 0 {
     614            2 :                         return errors.Newf("pebble: local ingestion sstables have overlapping ranges")
     615            2 :                 }
     616              :         }
     617            2 :         if len(lr.shared) == 0 {
     618            2 :                 return nil
     619            2 :         }
     620            0 :         filesInLevel := make([]*manifest.TableMetadata, 0, len(lr.shared))
     621            0 :         for l := sharedLevelsStart; l < numLevels; l++ {
     622            0 :                 filesInLevel = filesInLevel[:0]
     623            0 :                 for i := range lr.shared {
     624            0 :                         if lr.shared[i].shared.Level == uint8(l) {
     625            0 :                                 filesInLevel = append(filesInLevel, lr.shared[i].TableMetadata)
     626            0 :                         }
     627              :                 }
     628            0 :                 for i := range lr.external {
     629            0 :                         if lr.external[i].external.Level == uint8(l) {
     630            0 :                                 filesInLevel = append(filesInLevel, lr.external[i].TableMetadata)
     631            0 :                         }
     632              :                 }
     633            0 :                 slices.SortFunc(filesInLevel, func(a, b *manifest.TableMetadata) int {
     634            0 :                         return cmp(a.Smallest().UserKey, b.Smallest().UserKey)
     635            0 :                 })
     636            0 :                 for i := 1; i < len(filesInLevel); i++ {
     637            0 :                         if sstableKeyCompare(cmp, filesInLevel[i-1].Largest(), filesInLevel[i].Smallest()) >= 0 {
     638            0 :                                 return base.AssertionFailedf("pebble: external shared sstables have overlapping ranges")
     639            0 :                         }
     640              :                 }
     641              :         }
     642            0 :         return nil
     643              : }
     644              : 
     645            1 : func ingestCleanup(objProvider objstorage.Provider, meta []ingestLocalMeta) error {
     646            1 :         var firstErr error
     647            1 :         for i := range meta {
     648            1 :                 if err := objProvider.Remove(base.FileTypeTable, meta[i].TableBacking.DiskFileNum); err != nil {
     649            1 :                         firstErr = firstError(firstErr, err)
     650            1 :                 }
     651              :         }
     652            1 :         return firstErr
     653              : }
     654              : 
     655              : // ingestLinkLocal creates new objects which are backed by either hardlinks to or
     656              : // copies of the ingested files.
     657              : func ingestLinkLocal(
     658              :         ctx context.Context,
     659              :         jobID JobID,
     660              :         opts *Options,
     661              :         objProvider objstorage.Provider,
     662              :         localMetas []ingestLocalMeta,
     663            2 : ) error {
     664            2 :         for i := range localMetas {
     665            2 :                 objMeta, err := objProvider.LinkOrCopyFromLocal(
     666            2 :                         ctx, opts.FS, localMetas[i].path, base.FileTypeTable, localMetas[i].TableBacking.DiskFileNum,
     667            2 :                         objstorage.CreateOptions{PreferSharedStorage: true},
     668            2 :                 )
     669            2 :                 if err != nil {
     670            1 :                         if err2 := ingestCleanup(objProvider, localMetas[:i]); err2 != nil {
     671            0 :                                 opts.Logger.Errorf("ingest cleanup failed: %v", err2)
     672            0 :                         }
     673            1 :                         return err
     674              :                 }
     675            2 :                 if opts.EventListener.TableCreated != nil {
     676            2 :                         opts.EventListener.TableCreated(TableCreateInfo{
     677            2 :                                 JobID:   int(jobID),
     678            2 :                                 Reason:  "ingesting",
     679            2 :                                 Path:    objProvider.Path(objMeta),
     680            2 :                                 FileNum: base.PhysicalTableDiskFileNum(localMetas[i].TableNum),
     681            2 :                         })
     682            2 :                 }
     683              :         }
     684            2 :         return nil
     685              : }
     686              : 
     687              : // ingestAttachRemote attaches remote objects to the storage provider.
     688              : //
     689              : // For external objects, we reuse existing FileBackings from the current version
     690              : // when possible.
     691              : //
     692              : // ingestUnprotectExternalBackings() must be called after this function (even in
     693              : // error cases).
     694            2 : func (d *DB) ingestAttachRemote(jobID JobID, lr ingestLoadResult) error {
     695            2 :         remoteObjs := make([]objstorage.RemoteObjectToAttach, 0, len(lr.shared)+len(lr.external))
     696            2 :         for i := range lr.shared {
     697            2 :                 backing, err := lr.shared[i].shared.Backing.Get()
     698            2 :                 if err != nil {
     699            0 :                         return err
     700            0 :                 }
     701            2 :                 remoteObjs = append(remoteObjs, objstorage.RemoteObjectToAttach{
     702            2 :                         FileNum:  lr.shared[i].TableBacking.DiskFileNum,
     703            2 :                         FileType: base.FileTypeTable,
     704            2 :                         Backing:  backing,
     705            2 :                 })
     706              :         }
     707              : 
     708            2 :         d.findExistingBackingsForExternalObjects(lr.external)
     709            2 : 
     710            2 :         newTableBackings := make(map[remote.ObjectKey]*manifest.TableBacking, len(lr.external))
     711            2 :         for i := range lr.external {
     712            2 :                 meta := lr.external[i].TableMetadata
     713            2 :                 if meta.TableBacking != nil {
     714            2 :                         // The backing was filled in by findExistingBackingsForExternalObjects().
     715            2 :                         continue
     716              :                 }
     717            2 :                 key := remote.MakeObjectKey(lr.external[i].external.Locator, lr.external[i].external.ObjName)
     718            2 :                 if backing, ok := newTableBackings[key]; ok {
     719            2 :                         // We already created the same backing in this loop. Update its size.
     720            2 :                         backing.Size += lr.external[i].external.Size
     721            2 :                         meta.AttachVirtualBacking(backing)
     722            2 :                         continue
     723              :                 }
     724            2 :                 providerBacking, err := d.objProvider.CreateExternalObjectBacking(key.Locator, key.ObjectName)
     725            2 :                 if err != nil {
     726            0 :                         return err
     727            0 :                 }
     728              :                 // We have to attach the remote object (and assign it a DiskFileNum). For
     729              :                 // simplicity, we use the same number for both the FileNum and the
     730              :                 // DiskFileNum (even though this is a virtual sstable).
     731            2 :                 size := max(lr.external[i].external.Size, 1)
     732            2 :                 meta.InitVirtualBacking(base.DiskFileNum(meta.TableNum), size)
     733            2 : 
     734            2 :                 // Set the underlying TableBacking's size to the same size as the virtualized
     735            2 :                 // view of the sstable. This ensures that we don't over-prioritize this
     736            2 :                 // sstable for compaction just yet, as we do not have a clear sense of
     737            2 :                 // what parts of this sstable are referenced by other nodes.
     738            2 :                 meta.TableBacking.Size = size
     739            2 :                 newTableBackings[key] = meta.TableBacking
     740            2 : 
     741            2 :                 remoteObjs = append(remoteObjs, objstorage.RemoteObjectToAttach{
     742            2 :                         FileNum:  meta.TableBacking.DiskFileNum,
     743            2 :                         FileType: base.FileTypeTable,
     744            2 :                         Backing:  providerBacking,
     745            2 :                 })
     746              :         }
     747              : 
     748            2 :         for i := range lr.external {
     749            2 :                 if err := lr.external[i].Validate(d.opts.Comparer.Compare, d.opts.Comparer.FormatKey); err != nil {
     750            0 :                         return err
     751            0 :                 }
     752              :         }
     753              : 
     754            2 :         remoteObjMetas, err := d.objProvider.AttachRemoteObjects(remoteObjs)
     755            2 :         if err != nil {
     756            0 :                 return err
     757            0 :         }
     758              : 
     759            2 :         for i := range lr.shared {
     760            2 :                 // One corner case around file sizes we need to be mindful of, is that
     761            2 :                 // if one of the shareObjs was initially created by us (and has boomeranged
     762            2 :                 // back from another node), we'll need to update the TableBacking's size
     763            2 :                 // to be the true underlying size. Otherwise, we could hit errors when we
     764            2 :                 // open the db again after a crash/restart (see checkConsistency in open.go),
     765            2 :                 // plus it more accurately allows us to prioritize compactions of files
     766            2 :                 // that were originally created by us.
     767            2 :                 if remoteObjMetas[i].IsShared() && !d.objProvider.IsSharedForeign(remoteObjMetas[i]) {
     768            2 :                         size, err := d.objProvider.Size(remoteObjMetas[i])
     769            2 :                         if err != nil {
     770            0 :                                 return err
     771            0 :                         }
     772            2 :                         lr.shared[i].TableBacking.Size = max(uint64(size), 1)
     773              :                 }
     774              :         }
     775              : 
     776            2 :         if d.opts.EventListener.TableCreated != nil {
     777            2 :                 for i := range remoteObjMetas {
     778            2 :                         d.opts.EventListener.TableCreated(TableCreateInfo{
     779            2 :                                 JobID:   int(jobID),
     780            2 :                                 Reason:  "ingesting",
     781            2 :                                 Path:    d.objProvider.Path(remoteObjMetas[i]),
     782            2 :                                 FileNum: remoteObjMetas[i].DiskFileNum,
     783            2 :                         })
     784            2 :                 }
     785              :         }
     786              : 
     787            2 :         return nil
     788              : }
     789              : 
     790              : // findExistingBackingsForExternalObjects populates the TableBacking for external
     791              : // files which are already in use by the current version.
     792              : //
     793              : // We take a Ref and LatestRef on populated backings.
     794            2 : func (d *DB) findExistingBackingsForExternalObjects(metas []ingestExternalMeta) {
     795            2 :         d.mu.Lock()
     796            2 :         defer d.mu.Unlock()
     797            2 : 
     798            2 :         for i := range metas {
     799            2 :                 diskFileNums := d.objProvider.GetExternalObjects(metas[i].external.Locator, metas[i].external.ObjName)
     800            2 :                 // We cross-check against fileBackings in the current version because it is
     801            2 :                 // possible that the external object is referenced by an sstable which only
     802            2 :                 // exists in a previous version. In that case, that object could be removed
     803            2 :                 // at any time so we cannot reuse it.
     804            2 :                 for _, n := range diskFileNums {
     805            2 :                         if backing, ok := d.mu.versions.virtualBackings.Get(n); ok {
     806            2 :                                 // Protect this backing from being removed from the latest version. We
     807            2 :                                 // will unprotect in ingestUnprotectExternalBackings.
     808            2 :                                 d.mu.versions.virtualBackings.Protect(n)
     809            2 :                                 metas[i].usedExistingBacking = true
     810            2 :                                 metas[i].AttachVirtualBacking(backing)
     811            2 : 
     812            2 :                                 // We can't update the size of the backing here, so make sure the
     813            2 :                                 // virtual size is sane.
     814            2 :                                 // TODO(radu): investigate what would it take to update the backing size.
     815            2 :                                 metas[i].Size = min(metas[i].Size, backing.Size)
     816            2 :                                 break
     817              :                         }
     818              :                 }
     819              :         }
     820              : }
     821              : 
     822              : // ingestUnprotectExternalBackings unprotects the file backings that were reused
     823              : // for external objects when the ingestion fails.
     824            2 : func (d *DB) ingestUnprotectExternalBackings(lr ingestLoadResult) {
     825            2 :         d.mu.Lock()
     826            2 :         defer d.mu.Unlock()
     827            2 : 
     828            2 :         for _, meta := range lr.external {
     829            2 :                 if meta.usedExistingBacking {
     830            2 :                         // If the backing is not use anywhere else and the ingest failed (or the
     831            2 :                         // ingested tables were already compacted away), this call will cause in
     832            2 :                         // the next version update to remove the backing.
     833            2 :                         d.mu.versions.virtualBackings.Unprotect(meta.TableBacking.DiskFileNum)
     834            2 :                 }
     835              :         }
     836              : }
     837              : 
     838              : func setSeqNumInMetadata(
     839              :         m *manifest.TableMetadata, seqNum base.SeqNum, cmp Compare, format base.FormatKey,
     840            2 : ) error {
     841            2 :         setSeqFn := func(k base.InternalKey) base.InternalKey {
     842            2 :                 return base.MakeInternalKey(k.UserKey, seqNum, k.Kind())
     843            2 :         }
     844              :         // NB: we set the fields directly here, rather than via their Extend*
     845              :         // methods, as we are updating sequence numbers.
     846            2 :         if m.HasPointKeys {
     847            2 :                 m.PointKeyBounds.SetSmallest(setSeqFn(m.PointKeyBounds.Smallest()))
     848            2 :         }
     849            2 :         if m.HasRangeKeys {
     850            2 :                 m.RangeKeyBounds.SetSmallest(setSeqFn(m.RangeKeyBounds.Smallest()))
     851            2 :         }
     852              :         // Only update the seqnum for the largest key if that key is not an
     853              :         // "exclusive sentinel" (i.e. a range deletion sentinel or a range key
     854              :         // boundary), as doing so effectively drops the exclusive sentinel (by
     855              :         // lowering the seqnum from the max value), and extends the bounds of the
     856              :         // table.
     857              :         // NB: as the largest range key is always an exclusive sentinel, it is never
     858              :         // updated.
     859            2 :         if m.HasPointKeys && !m.PointKeyBounds.Largest().IsExclusiveSentinel() {
     860            2 :                 m.PointKeyBounds.SetLargest(setSeqFn(m.PointKeyBounds.Largest()))
     861            2 :         }
     862              :         // Setting smallestSeqNum == largestSeqNum triggers the setting of
     863              :         // Properties.GlobalSeqNum when an sstable is loaded.
     864            2 :         m.SmallestSeqNum = seqNum
     865            2 :         m.LargestSeqNum = seqNum
     866            2 :         m.LargestSeqNumAbsolute = seqNum
     867            2 :         // Ensure the new bounds are consistent.
     868            2 :         if err := m.Validate(cmp, format); err != nil {
     869            0 :                 return err
     870            0 :         }
     871            2 :         return nil
     872              : }
     873              : 
     874              : func ingestUpdateSeqNum(
     875              :         cmp Compare, format base.FormatKey, seqNum base.SeqNum, loadResult ingestLoadResult,
     876            2 : ) error {
     877            2 :         // Shared sstables are required to be sorted by level ascending. We then
     878            2 :         // iterate the shared sstables in reverse, assigning the lower sequence
     879            2 :         // numbers to the shared sstables that will be ingested into the lower
     880            2 :         // (larger numbered) levels first. This ensures sequence number shadowing is
     881            2 :         // correct.
     882            2 :         for i := len(loadResult.shared) - 1; i >= 0; i-- {
     883            2 :                 if i-1 >= 0 && loadResult.shared[i-1].shared.Level > loadResult.shared[i].shared.Level {
     884            0 :                         panic(errors.AssertionFailedf("shared files %s, %s out of order", loadResult.shared[i-1], loadResult.shared[i]))
     885              :                 }
     886            2 :                 if err := setSeqNumInMetadata(loadResult.shared[i].TableMetadata, seqNum, cmp, format); err != nil {
     887            0 :                         return err
     888            0 :                 }
     889            2 :                 seqNum++
     890              :         }
     891            2 :         for i := range loadResult.external {
     892            2 :                 if err := setSeqNumInMetadata(loadResult.external[i].TableMetadata, seqNum, cmp, format); err != nil {
     893            0 :                         return err
     894            0 :                 }
     895            2 :                 seqNum++
     896              :         }
     897            2 :         for i := range loadResult.local {
     898            2 :                 if err := setSeqNumInMetadata(loadResult.local[i].TableMetadata, seqNum, cmp, format); err != nil {
     899            0 :                         return err
     900            0 :                 }
     901            2 :                 seqNum++
     902              :         }
     903            2 :         return nil
     904              : }
     905              : 
     906              : // ingestTargetLevel returns the target level for a file being ingested.
     907              : // If suggestSplit is true, it accounts for ingest-time splitting as part of
     908              : // its target level calculation, and if a split candidate is found, that file
     909              : // is returned as the splitFile.
     910              : func ingestTargetLevel(
     911              :         ctx context.Context,
     912              :         cmp base.Compare,
     913              :         lsmOverlap overlap.WithLSM,
     914              :         baseLevel int,
     915              :         compactions map[*compaction]struct{},
     916              :         meta *manifest.TableMetadata,
     917              :         suggestSplit bool,
     918            2 : ) (targetLevel int, splitFile *manifest.TableMetadata, err error) {
     919            2 :         // Find the lowest level which does not have any files which overlap meta. We
     920            2 :         // search from L0 to L6 looking for whether there are any files in the level
     921            2 :         // which overlap meta. We want the "lowest" level (where lower means
     922            2 :         // increasing level number) in order to reduce write amplification.
     923            2 :         //
     924            2 :         // There are 2 kinds of overlap we need to check for: file boundary overlap
     925            2 :         // and data overlap. Data overlap implies file boundary overlap. Note that it
     926            2 :         // is always possible to ingest into L0.
     927            2 :         //
     928            2 :         // To place meta at level i where i > 0:
     929            2 :         // - there must not be any data overlap with levels <= i, since that will
     930            2 :         //   violate the sequence number invariant.
     931            2 :         // - no file boundary overlap with level i, since that will violate the
     932            2 :         //   invariant that files do not overlap in levels i > 0.
     933            2 :         //   - if there is only a file overlap at a given level, and no data overlap,
     934            2 :         //     we can still slot a file at that level. We return the fileMetadata with
     935            2 :         //     which we have file boundary overlap (must be only one file, as sstable
     936            2 :         //     bounds are usually tight on user keys) and the caller is expected to split
     937            2 :         //     that sstable into two virtual sstables, allowing this file to go into that
     938            2 :         //     level. Note that if we have file boundary overlap with two files, which
     939            2 :         //     should only happen on rare occasions, we treat it as data overlap and
     940            2 :         //     don't use this optimization.
     941            2 :         //
     942            2 :         // The file boundary overlap check is simpler to conceptualize. Consider the
     943            2 :         // following example, in which the ingested file lies completely before or
     944            2 :         // after the file being considered.
     945            2 :         //
     946            2 :         //   |--|           |--|  ingested file: [a,b] or [f,g]
     947            2 :         //         |-----|        existing file: [c,e]
     948            2 :         //  _____________________
     949            2 :         //   a  b  c  d  e  f  g
     950            2 :         //
     951            2 :         // In both cases the ingested file can move to considering the next level.
     952            2 :         //
     953            2 :         // File boundary overlap does not necessarily imply data overlap. The check
     954            2 :         // for data overlap is a little more nuanced. Consider the following examples:
     955            2 :         //
     956            2 :         //  1. No data overlap:
     957            2 :         //
     958            2 :         //          |-|   |--|    ingested file: [cc-d] or [ee-ff]
     959            2 :         //  |*--*--*----*------*| existing file: [a-g], points: [a, b, c, dd, g]
     960            2 :         //  _____________________
     961            2 :         //   a  b  c  d  e  f  g
     962            2 :         //
     963            2 :         // In this case the ingested files can "fall through" this level. The checks
     964            2 :         // continue at the next level.
     965            2 :         //
     966            2 :         //  2. Data overlap:
     967            2 :         //
     968            2 :         //            |--|        ingested file: [d-e]
     969            2 :         //  |*--*--*----*------*| existing file: [a-g], points: [a, b, c, dd, g]
     970            2 :         //  _____________________
     971            2 :         //   a  b  c  d  e  f  g
     972            2 :         //
     973            2 :         // In this case the file cannot be ingested into this level as the point 'dd'
     974            2 :         // is in the way.
     975            2 :         //
     976            2 :         // It is worth noting that the check for data overlap is only approximate. In
     977            2 :         // the previous example, the ingested table [d-e] could contain only the
     978            2 :         // points 'd' and 'e', in which case the table would be eligible for
     979            2 :         // considering lower levels. However, such a fine-grained check would need to
     980            2 :         // be exhaustive (comparing points and ranges in both the ingested existing
     981            2 :         // tables) and such a check is prohibitively expensive. Thus Pebble treats any
     982            2 :         // existing point that falls within the ingested table bounds as being "data
     983            2 :         // overlap".
     984            2 : 
     985            2 :         if lsmOverlap[0].Result == overlap.Data {
     986            2 :                 return 0, nil, nil
     987            2 :         }
     988            2 :         targetLevel = 0
     989            2 :         splitFile = nil
     990            2 :         for level := baseLevel; level < numLevels; level++ {
     991            2 :                 var candidateSplitFile *manifest.TableMetadata
     992            2 :                 switch lsmOverlap[level].Result {
     993            2 :                 case overlap.Data:
     994            2 :                         // We cannot ingest into or under this level; return the best target level
     995            2 :                         // so far.
     996            2 :                         return targetLevel, splitFile, nil
     997              : 
     998            2 :                 case overlap.OnlyBoundary:
     999            2 :                         if !suggestSplit || lsmOverlap[level].SplitFile == nil {
    1000            2 :                                 // We can ingest under this level, but not into this level.
    1001            2 :                                 continue
    1002              :                         }
    1003              :                         // We can ingest into this level if we split this file.
    1004            2 :                         candidateSplitFile = lsmOverlap[level].SplitFile
    1005              : 
    1006            2 :                 case overlap.None:
    1007              :                 // We can ingest into this level.
    1008              : 
    1009            0 :                 default:
    1010            0 :                         return 0, nil, base.AssertionFailedf("unexpected WithLevel.Result: %v", lsmOverlap[level].Result)
    1011              :                 }
    1012              : 
    1013              :                 // Check boundary overlap with any ongoing compactions. We consider an
    1014              :                 // overlapping compaction that's writing files to an output level as
    1015              :                 // equivalent to boundary overlap with files in that output level.
    1016              :                 //
    1017              :                 // We cannot check for data overlap with the new SSTs compaction will produce
    1018              :                 // since compaction hasn't been done yet. However, there's no need to check
    1019              :                 // since all keys in them will be from levels in [c.startLevel,
    1020              :                 // c.outputLevel], and all those levels have already had their data overlap
    1021              :                 // tested negative (else we'd have returned earlier).
    1022              :                 //
    1023              :                 // An alternative approach would be to cancel these compactions and proceed
    1024              :                 // with an ingest-time split on this level if necessary. However, compaction
    1025              :                 // cancellation can result in significant wasted effort and is best avoided
    1026              :                 // unless necessary.
    1027            2 :                 overlaps := false
    1028            2 :                 for c := range compactions {
    1029            2 :                         if c.outputLevel == nil || level != c.outputLevel.level {
    1030            2 :                                 continue
    1031              :                         }
    1032            2 :                         if cmp(meta.Smallest().UserKey, c.largest.UserKey) <= 0 &&
    1033            2 :                                 cmp(meta.Largest().UserKey, c.smallest.UserKey) >= 0 {
    1034            2 :                                 overlaps = true
    1035            2 :                                 break
    1036              :                         }
    1037              :                 }
    1038            2 :                 if !overlaps {
    1039            2 :                         targetLevel = level
    1040            2 :                         splitFile = candidateSplitFile
    1041            2 :                 }
    1042              :         }
    1043            2 :         return targetLevel, splitFile, nil
    1044              : }
    1045              : 
    1046              : // Ingest ingests a set of sstables into the DB. Ingestion of the files is
    1047              : // atomic and semantically equivalent to creating a single batch containing all
    1048              : // of the mutations in the sstables. Ingestion may require the memtable to be
    1049              : // flushed. The ingested sstable files are moved into the DB and must reside on
    1050              : // the same filesystem as the DB. Sstables can be created for ingestion using
    1051              : // sstable.Writer. On success, Ingest removes the input paths.
    1052              : //
    1053              : // Ingested sstables must have been created with a known KeySchema (when written
    1054              : // with columnar blocks) and Comparer. They must not contain any references to
    1055              : // external blob files.
    1056              : //
    1057              : // Two types of sstables are accepted for ingestion(s): one is sstables present
    1058              : // in the instance's vfs.FS and can be referenced locally. The other is sstables
    1059              : // present in remote.Storage, referred to as shared or foreign sstables. These
    1060              : // shared sstables can be linked through objstorageprovider.Provider, and do not
    1061              : // need to already be present on the local vfs.FS. Foreign sstables must all fit
    1062              : // in an excise span, and are destined for a level specified in SharedSSTMeta.
    1063              : //
    1064              : // All sstables *must* be Sync()'d by the caller after all bytes are written
    1065              : // and before its file handle is closed; failure to do so could violate
    1066              : // durability or lead to corrupted on-disk state. This method cannot, in a
    1067              : // platform-and-FS-agnostic way, ensure that all sstables in the input are
    1068              : // properly synced to disk. Opening new file handles and Sync()-ing them
    1069              : // does not always guarantee durability; see the discussion here on that:
    1070              : // https://github.com/cockroachdb/pebble/pull/835#issuecomment-663075379
    1071              : //
    1072              : // Ingestion loads each sstable into the lowest level of the LSM which it
    1073              : // doesn't overlap (see ingestTargetLevel). If an sstable overlaps a memtable,
    1074              : // ingestion forces the memtable to flush, and then waits for the flush to
    1075              : // occur. In some cases, such as with no foreign sstables and no excise span,
    1076              : // ingestion that gets blocked on a memtable can join the flushable queue and
    1077              : // finish even before the memtable has been flushed.
    1078              : //
    1079              : // The steps for ingestion are:
    1080              : //
    1081              : //  1. Allocate table numbers for every sstable being ingested.
    1082              : //  2. Load the metadata for all sstables being ingested.
    1083              : //  3. Sort the sstables by smallest key, verifying non overlap (for local
    1084              : //     sstables).
    1085              : //  4. Hard link (or copy) the local sstables into the DB directory.
    1086              : //  5. Allocate a sequence number to use for all of the entries in the
    1087              : //     local sstables. This is the step where overlap with memtables is
    1088              : //     determined. If there is overlap, we remember the most recent memtable
    1089              : //     that overlaps.
    1090              : //  6. Update the sequence number in the ingested local sstables. (Remote
    1091              : //     sstables get fixed sequence numbers that were determined at load time.)
    1092              : //  7. Wait for the most recent memtable that overlaps to flush (if any).
    1093              : //  8. Add the ingested sstables to the version (DB.ingestApply).
    1094              : //     8.1.  If an excise span was specified, figure out what sstables in the
    1095              : //     current version overlap with the excise span, and create new virtual
    1096              : //     sstables out of those sstables that exclude the excised span (DB.excise).
    1097              : //  9. Publish the ingestion sequence number.
    1098              : //
    1099              : // Note that if the mutable memtable overlaps with ingestion, a flush of the
    1100              : // memtable is forced equivalent to DB.Flush. Additionally, subsequent
    1101              : // mutations that get sequence numbers larger than the ingestion sequence
    1102              : // number get queued up behind the ingestion waiting for it to complete. This
    1103              : // can produce a noticeable hiccup in performance. See
    1104              : // https://github.com/cockroachdb/pebble/issues/25 for an idea for how to fix
    1105              : // this hiccup.
    1106            2 : func (d *DB) Ingest(ctx context.Context, paths []string) error {
    1107            2 :         if err := d.closed.Load(); err != nil {
    1108            1 :                 panic(err)
    1109              :         }
    1110            2 :         if d.opts.ReadOnly {
    1111            1 :                 return ErrReadOnly
    1112            1 :         }
    1113            2 :         _, err := d.ingest(ctx, ingestArgs{Local: paths})
    1114            2 :         return err
    1115              : }
    1116              : 
    1117              : // IngestOperationStats provides some information about where in the LSM the
    1118              : // bytes were ingested.
    1119              : type IngestOperationStats struct {
    1120              :         // Bytes is the total bytes in the ingested sstables.
    1121              :         Bytes uint64
    1122              :         // ApproxIngestedIntoL0Bytes is the approximate number of bytes ingested
    1123              :         // into L0. This value is approximate when flushable ingests are active and
    1124              :         // an ingest overlaps an entry in the flushable queue. Currently, this
    1125              :         // approximation is very rough, only including tables that overlapped the
    1126              :         // memtable. This estimate may be improved with #2112.
    1127              :         ApproxIngestedIntoL0Bytes uint64
    1128              :         // MemtableOverlappingFiles is the count of ingested sstables
    1129              :         // that overlapped keys in the memtables.
    1130              :         MemtableOverlappingFiles int
    1131              : }
    1132              : 
    1133              : // ExternalFile are external sstables that can be referenced through
    1134              : // objprovider and ingested as remote files that will not be refcounted or
    1135              : // cleaned up. For use with online restore. Note that the underlying sstable
    1136              : // could contain keys outside the [Smallest,Largest) bounds; however Pebble
    1137              : // is expected to only read the keys within those bounds.
    1138              : type ExternalFile struct {
    1139              :         // Locator is the shared.Locator that can be used with objProvider to
    1140              :         // resolve a reference to this external sstable.
    1141              :         Locator remote.Locator
    1142              : 
    1143              :         // ObjName is the unique name of this sstable on Locator.
    1144              :         ObjName string
    1145              : 
    1146              :         // Size of the referenced proportion of the virtualized sstable. An estimate
    1147              :         // is acceptable in lieu of the backing file size.
    1148              :         Size uint64
    1149              : 
    1150              :         // StartKey and EndKey define the bounds of the sstable; the ingestion
    1151              :         // of this file will only result in keys within [StartKey, EndKey) if
    1152              :         // EndKeyIsInclusive is false or [StartKey, EndKey] if it is true.
    1153              :         // These bounds are loose i.e. it's possible for keys to not span the
    1154              :         // entirety of this range.
    1155              :         //
    1156              :         // StartKey and EndKey user keys must not have suffixes.
    1157              :         //
    1158              :         // Multiple ExternalFiles in one ingestion must all have non-overlapping
    1159              :         // bounds.
    1160              :         StartKey, EndKey []byte
    1161              : 
    1162              :         // EndKeyIsInclusive is true if EndKey should be treated as inclusive.
    1163              :         EndKeyIsInclusive bool
    1164              : 
    1165              :         // HasPointKey and HasRangeKey denote whether this file contains point keys
    1166              :         // or range keys. If both structs are false, an error is returned during
    1167              :         // ingestion.
    1168              :         HasPointKey, HasRangeKey bool
    1169              : 
    1170              :         // SyntheticPrefix will prepend this suffix to all keys in the file during
    1171              :         // iteration. Note that the backing file itself is not modified.
    1172              :         //
    1173              :         // SyntheticPrefix must be a prefix of both Bounds.Start and Bounds.End.
    1174              :         SyntheticPrefix []byte
    1175              : 
    1176              :         // SyntheticSuffix will replace the suffix of every key in the file during
    1177              :         // iteration. Note that the file itself is not modified, rather, every key
    1178              :         // returned by an iterator will have the synthetic suffix.
    1179              :         //
    1180              :         // SyntheticSuffix can only be used under the following conditions:
    1181              :         //  - the synthetic suffix must sort before any non-empty suffixes in the
    1182              :         //    backing sst (the entire sst, not just the part restricted to Bounds).
    1183              :         //  - the backing sst must not contain multiple keys with the same prefix.
    1184              :         SyntheticSuffix []byte
    1185              : 
    1186              :         // Level denotes the level at which this file was present at read time
    1187              :         // if the external file was returned by a scan of an existing Pebble
    1188              :         // instance. If Level is 0, this field is ignored.
    1189              :         Level uint8
    1190              : }
    1191              : 
    1192              : // IngestWithStats does the same as Ingest, and additionally returns
    1193              : // IngestOperationStats.
    1194            1 : func (d *DB) IngestWithStats(ctx context.Context, paths []string) (IngestOperationStats, error) {
    1195            1 :         if err := d.closed.Load(); err != nil {
    1196            0 :                 panic(err)
    1197              :         }
    1198            1 :         if d.opts.ReadOnly {
    1199            0 :                 return IngestOperationStats{}, ErrReadOnly
    1200            0 :         }
    1201            1 :         return d.ingest(ctx, ingestArgs{Local: paths})
    1202              : }
    1203              : 
    1204              : // IngestExternalFiles does the same as IngestWithStats, and additionally
    1205              : // accepts external files (with locator info that can be resolved using
    1206              : // d.opts.SharedStorage). These files must also be non-overlapping with
    1207              : // each other, and must be resolvable through d.objProvider.
    1208              : func (d *DB) IngestExternalFiles(
    1209              :         ctx context.Context, external []ExternalFile,
    1210            2 : ) (IngestOperationStats, error) {
    1211            2 :         if err := d.closed.Load(); err != nil {
    1212            0 :                 panic(err)
    1213              :         }
    1214              : 
    1215            2 :         if d.opts.ReadOnly {
    1216            0 :                 return IngestOperationStats{}, ErrReadOnly
    1217            0 :         }
    1218            2 :         if d.opts.Experimental.RemoteStorage == nil {
    1219            0 :                 return IngestOperationStats{}, errors.New("pebble: cannot ingest external files without shared storage configured")
    1220            0 :         }
    1221            2 :         return d.ingest(ctx, ingestArgs{External: external})
    1222              : }
    1223              : 
    1224              : // IngestAndExcise does the same as IngestWithStats, and additionally accepts a
    1225              : // list of shared files to ingest that can be read from a remote.Storage through
    1226              : // a Provider. All the shared files must live within exciseSpan, and any existing
    1227              : // keys in exciseSpan are deleted by turning existing sstables into virtual
    1228              : // sstables (if not virtual already) and shrinking their spans to exclude
    1229              : // exciseSpan. See the comment at Ingest for a more complete picture of the
    1230              : // ingestion process.
    1231              : //
    1232              : // Panics if this DB instance was not instantiated with a remote.Storage and
    1233              : // shared sstables are present.
    1234              : func (d *DB) IngestAndExcise(
    1235              :         ctx context.Context,
    1236              :         paths []string,
    1237              :         shared []SharedSSTMeta,
    1238              :         external []ExternalFile,
    1239              :         exciseSpan KeyRange,
    1240            2 : ) (IngestOperationStats, error) {
    1241            2 :         if err := d.closed.Load(); err != nil {
    1242            0 :                 panic(err)
    1243              :         }
    1244            2 :         if d.opts.ReadOnly {
    1245            0 :                 return IngestOperationStats{}, ErrReadOnly
    1246            0 :         }
    1247              :         // Excise is only supported on prefix keys.
    1248            2 :         if d.opts.Comparer.Split(exciseSpan.Start) != len(exciseSpan.Start) {
    1249            0 :                 return IngestOperationStats{}, errors.New("IngestAndExcise called with suffixed start key")
    1250            0 :         }
    1251            2 :         if d.opts.Comparer.Split(exciseSpan.End) != len(exciseSpan.End) {
    1252            0 :                 return IngestOperationStats{}, errors.New("IngestAndExcise called with suffixed end key")
    1253            0 :         }
    1254            2 :         if v := d.FormatMajorVersion(); v < FormatMinForSharedObjects {
    1255            0 :                 return IngestOperationStats{}, errors.Newf(
    1256            0 :                         "store has format major version %d; IngestAndExcise requires at least %d",
    1257            0 :                         v, FormatMinForSharedObjects,
    1258            0 :                 )
    1259            0 :         }
    1260            2 :         args := ingestArgs{
    1261            2 :                 Local:              paths,
    1262            2 :                 Shared:             shared,
    1263            2 :                 External:           external,
    1264            2 :                 ExciseSpan:         exciseSpan,
    1265            2 :                 ExciseBoundsPolicy: tightExciseBounds,
    1266            2 :         }
    1267            2 :         return d.ingest(ctx, args)
    1268              : }
    1269              : 
    1270              : // Both DB.mu and commitPipeline.mu must be held while this is called.
    1271              : func (d *DB) newIngestedFlushableEntry(
    1272              :         meta []*manifest.TableMetadata, seqNum base.SeqNum, logNum base.DiskFileNum, exciseSpan KeyRange,
    1273            2 : ) (*flushableEntry, error) {
    1274            2 :         // If there's an excise being done atomically with the same ingest, we
    1275            2 :         // assign the lowest sequence number in the set of sequence numbers for this
    1276            2 :         // ingestion to the excise. Note that we've already allocated fileCount+1
    1277            2 :         // sequence numbers in this case.
    1278            2 :         //
    1279            2 :         // This mimics the behaviour in the non-flushable ingest case (see the callsite
    1280            2 :         // for ingestUpdateSeqNum).
    1281            2 :         fileSeqNumStart := seqNum
    1282            2 :         if exciseSpan.Valid() {
    1283            2 :                 fileSeqNumStart = seqNum + 1 // the first seqNum is reserved for the excise.
    1284            2 :                 // The excise span will be retained by the flushable, outliving the
    1285            2 :                 // caller's ingestion call. Copy it.
    1286            2 :                 exciseSpan = KeyRange{
    1287            2 :                         Start: slices.Clone(exciseSpan.Start),
    1288            2 :                         End:   slices.Clone(exciseSpan.End),
    1289            2 :                 }
    1290            2 :         }
    1291              :         // Update the sequence number for all of the sstables in the
    1292              :         // metadata. Writing the metadata to the manifest when the
    1293              :         // version edit is applied is the mechanism that persists the
    1294              :         // sequence number. The sstables themselves are left unmodified.
    1295              :         // In this case, a version edit will only be written to the manifest
    1296              :         // when the flushable is eventually flushed. If Pebble restarts in that
    1297              :         // time, then we'll lose the ingest sequence number information. But this
    1298              :         // information will also be reconstructed on node restart.
    1299            2 :         for i, m := range meta {
    1300            2 :                 if err := setSeqNumInMetadata(m, fileSeqNumStart+base.SeqNum(i), d.cmp, d.opts.Comparer.FormatKey); err != nil {
    1301            0 :                         return nil, err
    1302            0 :                 }
    1303              :         }
    1304              : 
    1305            2 :         f := newIngestedFlushable(meta, d.opts.Comparer, d.newIters, d.tableNewRangeKeyIter, exciseSpan, seqNum)
    1306            2 : 
    1307            2 :         // NB: The logNum/seqNum are the WAL number which we're writing this entry
    1308            2 :         // to and the sequence number within the WAL which we'll write this entry
    1309            2 :         // to.
    1310            2 :         entry := d.newFlushableEntry(f, logNum, seqNum)
    1311            2 :         // The flushable entry starts off with a single reader ref, so increment
    1312            2 :         // the TableMetadata.Refs.
    1313            2 :         for _, file := range f.files {
    1314            2 :                 file.Ref()
    1315            2 :         }
    1316            2 :         entry.unrefFiles = func(of *manifest.ObsoleteFiles) {
    1317            2 :                 // Invoke Unref on each table. If any files become obsolete, they'll be
    1318            2 :                 // added to the set of obsolete files.
    1319            2 :                 for _, file := range f.files {
    1320            2 :                         file.Unref(of)
    1321            2 :                 }
    1322              :         }
    1323              : 
    1324            2 :         entry.flushForced = true
    1325            2 :         entry.releaseMemAccounting = func() {}
    1326            2 :         return entry, nil
    1327              : }
    1328              : 
    1329              : // Both DB.mu and commitPipeline.mu must be held while this is called. Since
    1330              : // we're holding both locks, the order in which we rotate the memtable or
    1331              : // recycle the WAL in this function is irrelevant as long as the correct log
    1332              : // numbers are assigned to the appropriate flushable.
    1333              : func (d *DB) handleIngestAsFlushable(
    1334              :         meta []*manifest.TableMetadata, seqNum base.SeqNum, exciseSpan KeyRange,
    1335            2 : ) error {
    1336            2 :         b := d.NewBatch()
    1337            2 :         if exciseSpan.Valid() {
    1338            2 :                 b.excise(exciseSpan.Start, exciseSpan.End)
    1339            2 :         }
    1340            2 :         for _, m := range meta {
    1341            2 :                 b.ingestSST(m.TableNum)
    1342            2 :         }
    1343            2 :         b.setSeqNum(seqNum)
    1344            2 : 
    1345            2 :         // If the WAL is disabled, then the logNum used to create the flushable
    1346            2 :         // entry doesn't matter. We just use the logNum assigned to the current
    1347            2 :         // mutable memtable. If the WAL is enabled, then this logNum will be
    1348            2 :         // overwritten by the logNum of the log which will contain the log entry
    1349            2 :         // for the ingestedFlushable.
    1350            2 :         logNum := d.mu.mem.queue[len(d.mu.mem.queue)-1].logNum
    1351            2 :         if !d.opts.DisableWAL {
    1352            2 :                 // We create a new WAL for the flushable instead of reusing the end of
    1353            2 :                 // the previous WAL. This simplifies the increment of the minimum
    1354            2 :                 // unflushed log number, and also simplifies WAL replay.
    1355            2 :                 var prevLogSize uint64
    1356            2 :                 logNum, prevLogSize = d.rotateWAL()
    1357            2 :                 // As the rotator of the WAL, we're responsible for updating the
    1358            2 :                 // previous flushable queue tail's log size.
    1359            2 :                 d.mu.mem.queue[len(d.mu.mem.queue)-1].logSize = prevLogSize
    1360            2 : 
    1361            2 :                 d.mu.Unlock()
    1362            2 :                 err := d.commit.directWrite(b)
    1363            2 :                 if err != nil {
    1364            0 :                         d.opts.Logger.Fatalf("%v", err)
    1365            0 :                 }
    1366            2 :                 d.mu.Lock()
    1367              :         }
    1368              : 
    1369            2 :         entry, err := d.newIngestedFlushableEntry(meta, seqNum, logNum, exciseSpan)
    1370            2 :         if err != nil {
    1371            0 :                 return err
    1372            0 :         }
    1373            2 :         nextSeqNum := seqNum + base.SeqNum(b.Count())
    1374            2 : 
    1375            2 :         // Set newLogNum to the logNum of the previous flushable. This value is
    1376            2 :         // irrelevant if the WAL is disabled. If the WAL is enabled, then we set
    1377            2 :         // the appropriate value below.
    1378            2 :         newLogNum := d.mu.mem.queue[len(d.mu.mem.queue)-1].logNum
    1379            2 :         if !d.opts.DisableWAL {
    1380            2 :                 // newLogNum will be the WAL num of the next mutable memtable which
    1381            2 :                 // comes after the ingestedFlushable in the flushable queue. The mutable
    1382            2 :                 // memtable will be created below.
    1383            2 :                 //
    1384            2 :                 // The prevLogSize returned by rotateWAL is the WAL to which the
    1385            2 :                 // flushable ingest keys were appended. This intermediary WAL is only
    1386            2 :                 // used to record the flushable ingest and nothing else.
    1387            2 :                 newLogNum, entry.logSize = d.rotateWAL()
    1388            2 :         }
    1389              : 
    1390            2 :         d.mu.versions.metrics.Ingest.Count++
    1391            2 :         currMem := d.mu.mem.mutable
    1392            2 :         // NB: Placing ingested sstables above the current memtables
    1393            2 :         // requires rotating of the existing memtables/WAL. There is
    1394            2 :         // some concern of churning through tiny memtables due to
    1395            2 :         // ingested sstables being placed on top of them, but those
    1396            2 :         // memtables would have to be flushed anyways.
    1397            2 :         d.mu.mem.queue = append(d.mu.mem.queue, entry)
    1398            2 :         d.rotateMemtable(newLogNum, nextSeqNum, currMem, 0 /* minSize */)
    1399            2 :         d.updateReadStateLocked(d.opts.DebugCheck)
    1400            2 :         // TODO(aaditya): is this necessary? we call this already in rotateMemtable above
    1401            2 :         d.maybeScheduleFlush()
    1402            2 :         return nil
    1403              : }
    1404              : 
    1405              : type ingestArgs struct {
    1406              :         // Local sstables to ingest.
    1407              :         Local []string
    1408              :         // Shared sstables to ingest.
    1409              :         Shared []SharedSSTMeta
    1410              :         // External sstables to ingest.
    1411              :         External []ExternalFile
    1412              :         // ExciseSpan (unset if not excising).
    1413              :         ExciseSpan         KeyRange
    1414              :         ExciseBoundsPolicy exciseBoundsPolicy
    1415              : }
    1416              : 
    1417              : // See comment at Ingest() for details on how this works.
    1418            2 : func (d *DB) ingest(ctx context.Context, args ingestArgs) (IngestOperationStats, error) {
    1419            2 :         paths := args.Local
    1420            2 :         shared := args.Shared
    1421            2 :         external := args.External
    1422            2 :         if len(shared) > 0 && d.opts.Experimental.RemoteStorage == nil {
    1423            0 :                 panic("cannot ingest shared sstables with nil SharedStorage")
    1424              :         }
    1425            2 :         if (args.ExciseSpan.Valid() || len(shared) > 0 || len(external) > 0) && d.FormatMajorVersion() < FormatVirtualSSTables {
    1426            0 :                 return IngestOperationStats{}, errors.New("pebble: format major version too old for excise, shared or external sstable ingestion")
    1427            0 :         }
    1428            2 :         if len(external) > 0 && d.FormatMajorVersion() < FormatSyntheticPrefixSuffix {
    1429            1 :                 for i := range external {
    1430            1 :                         if len(external[i].SyntheticPrefix) > 0 {
    1431            1 :                                 return IngestOperationStats{}, errors.New("pebble: format major version too old for synthetic prefix ingestion")
    1432            1 :                         }
    1433            1 :                         if len(external[i].SyntheticSuffix) > 0 {
    1434            1 :                                 return IngestOperationStats{}, errors.New("pebble: format major version too old for synthetic suffix ingestion")
    1435            1 :                         }
    1436              :                 }
    1437              :         }
    1438              :         // Allocate table numbers for all files being ingested and mark them as
    1439              :         // pending in order to prevent them from being deleted. Note that this causes
    1440              :         // the file number ordering to be out of alignment with sequence number
    1441              :         // ordering. The sorting of L0 tables by sequence number avoids relying on
    1442              :         // that (busted) invariant.
    1443            2 :         pendingOutputs := make([]base.TableNum, len(paths)+len(shared)+len(external))
    1444            2 :         for i := 0; i < len(paths)+len(shared)+len(external); i++ {
    1445            2 :                 pendingOutputs[i] = d.mu.versions.getNextTableNum()
    1446            2 :         }
    1447              : 
    1448            2 :         jobID := d.newJobID()
    1449            2 : 
    1450            2 :         // Load the metadata for all the files being ingested. This step detects
    1451            2 :         // and elides empty sstables.
    1452            2 :         loadResult, err := ingestLoad(ctx, d.opts, d.FormatMajorVersion(), paths, shared, external, d.cacheHandle, pendingOutputs)
    1453            2 :         if err != nil {
    1454            1 :                 return IngestOperationStats{}, err
    1455            1 :         }
    1456              : 
    1457            2 :         if loadResult.fileCount() == 0 && !args.ExciseSpan.Valid() {
    1458            2 :                 // All of the sstables to be ingested were empty. Nothing to do.
    1459            2 :                 return IngestOperationStats{}, nil
    1460            2 :         }
    1461              : 
    1462              :         // Verify the sstables do not overlap.
    1463            2 :         if err := ingestSortAndVerify(d.cmp, loadResult, args.ExciseSpan); err != nil {
    1464            2 :                 return IngestOperationStats{}, err
    1465            2 :         }
    1466              : 
    1467              :         // Hard link the sstables into the DB directory. Since the sstables aren't
    1468              :         // referenced by a version, they won't be used. If the hard linking fails
    1469              :         // (e.g. because the files reside on a different filesystem), ingestLinkLocal
    1470              :         // will fall back to copying, and if that fails we undo our work and return an
    1471              :         // error.
    1472            2 :         if err := ingestLinkLocal(ctx, jobID, d.opts, d.objProvider, loadResult.local); err != nil {
    1473            0 :                 return IngestOperationStats{}, err
    1474            0 :         }
    1475              : 
    1476            2 :         err = d.ingestAttachRemote(jobID, loadResult)
    1477            2 :         defer d.ingestUnprotectExternalBackings(loadResult)
    1478            2 :         if err != nil {
    1479            0 :                 return IngestOperationStats{}, err
    1480            0 :         }
    1481              : 
    1482              :         // Make the new tables durable. We need to do this at some point before we
    1483              :         // update the MANIFEST (via UpdateVersionLocked), otherwise a crash can have
    1484              :         // the tables referenced in the MANIFEST, but not present in the provider.
    1485            2 :         if err := d.objProvider.Sync(); err != nil {
    1486            1 :                 return IngestOperationStats{}, err
    1487            1 :         }
    1488              : 
    1489              :         // metaFlushableOverlaps is a map indicating which of the ingested sstables
    1490              :         // overlap some table in the flushable queue. It's used to approximate
    1491              :         // ingest-into-L0 stats when using flushable ingests.
    1492            2 :         metaFlushableOverlaps := make(map[base.TableNum]bool, loadResult.fileCount())
    1493            2 :         var mem *flushableEntry
    1494            2 :         var mut *memTable
    1495            2 :         // asFlushable indicates whether the sstable was ingested as a flushable.
    1496            2 :         var asFlushable bool
    1497            2 :         prepare := func(seqNum base.SeqNum) {
    1498            2 :                 // Note that d.commit.mu is held by commitPipeline when calling prepare.
    1499            2 : 
    1500            2 :                 // Determine the set of bounds we care about for the purpose of checking
    1501            2 :                 // for overlap among the flushables. If there's an excise span, we need
    1502            2 :                 // to check for overlap with its bounds as well.
    1503            2 :                 overlapBounds := make([]bounded, 0, loadResult.fileCount()+1)
    1504            2 :                 for _, m := range loadResult.local {
    1505            2 :                         overlapBounds = append(overlapBounds, m.TableMetadata)
    1506            2 :                 }
    1507            2 :                 for _, m := range loadResult.shared {
    1508            2 :                         overlapBounds = append(overlapBounds, m.TableMetadata)
    1509            2 :                 }
    1510            2 :                 for _, m := range loadResult.external {
    1511            2 :                         overlapBounds = append(overlapBounds, m.TableMetadata)
    1512            2 :                 }
    1513            2 :                 if args.ExciseSpan.Valid() {
    1514            2 :                         overlapBounds = append(overlapBounds, &args.ExciseSpan)
    1515            2 :                 }
    1516              : 
    1517            2 :                 d.mu.Lock()
    1518            2 :                 defer d.mu.Unlock()
    1519            2 : 
    1520            2 :                 if args.ExciseSpan.Valid() {
    1521            2 :                         // Check if any of the currently-open EventuallyFileOnlySnapshots
    1522            2 :                         // overlap in key ranges with the excise span. If so, we need to
    1523            2 :                         // check for memtable overlaps with all bounds of that
    1524            2 :                         // EventuallyFileOnlySnapshot in addition to the ingestion's own
    1525            2 :                         // bounds too.
    1526            2 :                         overlapBounds = append(overlapBounds, exciseOverlapBounds(
    1527            2 :                                 d.cmp, &d.mu.snapshots.snapshotList, args.ExciseSpan, seqNum)...)
    1528            2 :                 }
    1529              : 
    1530              :                 // Check to see if any files overlap with any of the memtables. The queue
    1531              :                 // is ordered from oldest to newest with the mutable memtable being the
    1532              :                 // last element in the slice. We want to wait for the newest table that
    1533              :                 // overlaps.
    1534              : 
    1535            2 :                 for i := len(d.mu.mem.queue) - 1; i >= 0; i-- {
    1536            2 :                         m := d.mu.mem.queue[i]
    1537            2 :                         m.computePossibleOverlaps(func(b bounded) shouldContinue {
    1538            2 :                                 // If this is the first table to overlap a flushable, save
    1539            2 :                                 // the flushable. This ingest must be ingested or flushed
    1540            2 :                                 // after it.
    1541            2 :                                 if mem == nil {
    1542            2 :                                         mem = m
    1543            2 :                                 }
    1544              : 
    1545            2 :                                 switch v := b.(type) {
    1546            2 :                                 case *manifest.TableMetadata:
    1547            2 :                                         // NB: False positives are possible if `m` is a flushable
    1548            2 :                                         // ingest that overlaps the file `v` in bounds but doesn't
    1549            2 :                                         // contain overlapping data. This is considered acceptable
    1550            2 :                                         // because it's rare (in CockroachDB a bound overlap likely
    1551            2 :                                         // indicates a data overlap), and blocking the commit
    1552            2 :                                         // pipeline while we perform I/O to check for overlap may be
    1553            2 :                                         // more disruptive than enqueueing this ingestion on the
    1554            2 :                                         // flushable queue and switching to a new memtable.
    1555            2 :                                         metaFlushableOverlaps[v.TableNum] = true
    1556            2 :                                 case *KeyRange:
    1557              :                                         // An excise span or an EventuallyFileOnlySnapshot protected range;
    1558              :                                         // not a file.
    1559            0 :                                 default:
    1560            0 :                                         panic("unreachable")
    1561              :                                 }
    1562            2 :                                 return continueIteration
    1563              :                         }, overlapBounds...)
    1564              :                 }
    1565              : 
    1566            2 :                 if mem == nil {
    1567            2 :                         // No overlap with any of the queued flushables, so no need to queue
    1568            2 :                         // after them.
    1569            2 : 
    1570            2 :                         // New writes with higher sequence numbers may be concurrently
    1571            2 :                         // committed. We must ensure they don't flush before this ingest
    1572            2 :                         // completes. To do that, we ref the mutable memtable as a writer,
    1573            2 :                         // preventing its flushing (and the flushing of all subsequent
    1574            2 :                         // flushables in the queue). Once we've acquired the manifest lock
    1575            2 :                         // to add the ingested sstables to the LSM, we can unref as we're
    1576            2 :                         // guaranteed that the flush won't edit the LSM before this ingest.
    1577            2 :                         mut = d.mu.mem.mutable
    1578            2 :                         mut.writerRef()
    1579            2 :                         return
    1580            2 :                 }
    1581              : 
    1582              :                 // The ingestion overlaps with some entry in the flushable queue. If the
    1583              :                 // pre-conditions are met below, we can treat this ingestion as a flushable
    1584              :                 // ingest, otherwise we wait on the memtable flush before ingestion.
    1585              :                 //
    1586              :                 // TODO(aaditya): We should make flushableIngest compatible with remote
    1587              :                 // files.
    1588            2 :                 hasRemoteFiles := len(shared) > 0 || len(external) > 0
    1589            2 :                 canIngestFlushable := d.FormatMajorVersion() >= FormatFlushableIngest &&
    1590            2 :                         (len(d.mu.mem.queue) < d.opts.MemTableStopWritesThreshold) &&
    1591            2 :                         !d.opts.Experimental.DisableIngestAsFlushable() && !hasRemoteFiles &&
    1592            2 :                         (!args.ExciseSpan.Valid() || d.FormatMajorVersion() >= FormatFlushableIngestExcises)
    1593            2 : 
    1594            2 :                 if !canIngestFlushable {
    1595            2 :                         // We're not able to ingest as a flushable,
    1596            2 :                         // so we must synchronously flush.
    1597            2 :                         //
    1598            2 :                         // TODO(bilal): Currently, if any of the files being ingested are shared,
    1599            2 :                         // we cannot use flushable ingests and need
    1600            2 :                         // to wait synchronously.
    1601            2 :                         if mem.flushable == d.mu.mem.mutable {
    1602            2 :                                 err = d.makeRoomForWrite(nil)
    1603            2 :                         }
    1604              :                         // New writes with higher sequence numbers may be concurrently
    1605              :                         // committed. We must ensure they don't flush before this ingest
    1606              :                         // completes. To do that, we ref the mutable memtable as a writer,
    1607              :                         // preventing its flushing (and the flushing of all subsequent
    1608              :                         // flushables in the queue). Once we've acquired the manifest lock
    1609              :                         // to add the ingested sstables to the LSM, we can unref as we're
    1610              :                         // guaranteed that the flush won't edit the LSM before this ingest.
    1611            2 :                         mut = d.mu.mem.mutable
    1612            2 :                         mut.writerRef()
    1613            2 :                         mem.flushForced = true
    1614            2 :                         d.maybeScheduleFlush()
    1615            2 :                         return
    1616              :                 }
    1617              :                 // Since there aren't too many memtables already queued up, we can
    1618              :                 // slide the ingested sstables on top of the existing memtables.
    1619            2 :                 asFlushable = true
    1620            2 :                 fileMetas := make([]*manifest.TableMetadata, len(loadResult.local))
    1621            2 :                 for i := range fileMetas {
    1622            2 :                         fileMetas[i] = loadResult.local[i].TableMetadata
    1623            2 :                 }
    1624            2 :                 err = d.handleIngestAsFlushable(fileMetas, seqNum, args.ExciseSpan)
    1625              :         }
    1626              : 
    1627            2 :         var ve *manifest.VersionEdit
    1628            2 :         apply := func(seqNum base.SeqNum) {
    1629            2 :                 if err != nil || asFlushable {
    1630            2 :                         // An error occurred during prepare.
    1631            2 :                         if mut != nil {
    1632            0 :                                 if mut.writerUnref() {
    1633            0 :                                         d.mu.Lock()
    1634            0 :                                         d.maybeScheduleFlush()
    1635            0 :                                         d.mu.Unlock()
    1636            0 :                                 }
    1637              :                         }
    1638            2 :                         return
    1639              :                 }
    1640              : 
    1641              :                 // If there's an excise being done atomically with the same ingest, we
    1642              :                 // assign the lowest sequence number in the set of sequence numbers for this
    1643              :                 // ingestion to the excise. Note that we've already allocated fileCount+1
    1644              :                 // sequence numbers in this case.
    1645            2 :                 if args.ExciseSpan.Valid() {
    1646            2 :                         seqNum++ // the first seqNum is reserved for the excise.
    1647            2 :                 }
    1648              :                 // Update the sequence numbers for all ingested sstables'
    1649              :                 // metadata. When the version edit is applied, the metadata is
    1650              :                 // written to the manifest, persisting the sequence number.
    1651              :                 // The sstables themselves are left unmodified.
    1652            2 :                 if err = ingestUpdateSeqNum(
    1653            2 :                         d.cmp, d.opts.Comparer.FormatKey, seqNum, loadResult,
    1654            2 :                 ); err != nil {
    1655            0 :                         if mut != nil {
    1656            0 :                                 if mut.writerUnref() {
    1657            0 :                                         d.mu.Lock()
    1658            0 :                                         d.maybeScheduleFlush()
    1659            0 :                                         d.mu.Unlock()
    1660            0 :                                 }
    1661              :                         }
    1662            0 :                         return
    1663              :                 }
    1664              : 
    1665              :                 // If we overlapped with a memtable in prepare wait for the flush to
    1666              :                 // finish.
    1667            2 :                 if mem != nil {
    1668            2 :                         <-mem.flushed
    1669            2 :                 }
    1670              : 
    1671              :                 // Assign the sstables to the correct level in the LSM and apply the
    1672              :                 // version edit.
    1673            2 :                 ve, err = d.ingestApply(ctx, jobID, loadResult, mut, args.ExciseSpan, args.ExciseBoundsPolicy, seqNum)
    1674              :         }
    1675              : 
    1676              :         // Only one ingest can occur at a time because if not, one would block waiting
    1677              :         // for the other to finish applying. This blocking would happen while holding
    1678              :         // the commit mutex which would prevent unrelated batches from writing their
    1679              :         // changes to the WAL and memtable. This will cause a bigger commit hiccup
    1680              :         // during ingestion.
    1681            2 :         seqNumCount := loadResult.fileCount()
    1682            2 :         if args.ExciseSpan.Valid() {
    1683            2 :                 seqNumCount++
    1684            2 :         }
    1685            2 :         d.commit.ingestSem <- struct{}{}
    1686            2 :         d.commit.AllocateSeqNum(seqNumCount, prepare, apply)
    1687            2 :         <-d.commit.ingestSem
    1688            2 : 
    1689            2 :         if err != nil {
    1690            1 :                 if err2 := ingestCleanup(d.objProvider, loadResult.local); err2 != nil {
    1691            0 :                         d.opts.Logger.Errorf("ingest cleanup failed: %v", err2)
    1692            0 :                 }
    1693            2 :         } else {
    1694            2 :                 // Since we either created a hard link to the ingesting files, or copied
    1695            2 :                 // them over, it is safe to remove the originals paths.
    1696            2 :                 for i := range loadResult.local {
    1697            2 :                         path := loadResult.local[i].path
    1698            2 :                         if err2 := d.opts.FS.Remove(path); err2 != nil {
    1699            1 :                                 d.opts.Logger.Errorf("ingest failed to remove original file: %s", err2)
    1700            1 :                         }
    1701              :                 }
    1702              :         }
    1703              : 
    1704              :         // TODO(jackson): Refactor this so that the case where there are no files
    1705              :         // but a valid excise span is not so exceptional.
    1706              : 
    1707            2 :         var stats IngestOperationStats
    1708            2 :         if loadResult.fileCount() > 0 {
    1709            2 :                 info := TableIngestInfo{
    1710            2 :                         JobID:     int(jobID),
    1711            2 :                         Err:       err,
    1712            2 :                         flushable: asFlushable,
    1713            2 :                 }
    1714            2 :                 if len(loadResult.local) > 0 {
    1715            2 :                         info.GlobalSeqNum = loadResult.local[0].SmallestSeqNum
    1716            2 :                 } else if len(loadResult.shared) > 0 {
    1717            2 :                         info.GlobalSeqNum = loadResult.shared[0].SmallestSeqNum
    1718            2 :                 } else {
    1719            2 :                         info.GlobalSeqNum = loadResult.external[0].SmallestSeqNum
    1720            2 :                 }
    1721            2 :                 if ve != nil {
    1722            2 :                         info.Tables = make([]struct {
    1723            2 :                                 TableInfo
    1724            2 :                                 Level int
    1725            2 :                         }, len(ve.NewTables))
    1726            2 :                         for i := range ve.NewTables {
    1727            2 :                                 e := &ve.NewTables[i]
    1728            2 :                                 info.Tables[i].Level = e.Level
    1729            2 :                                 info.Tables[i].TableInfo = e.Meta.TableInfo()
    1730            2 :                                 stats.Bytes += e.Meta.Size
    1731            2 :                                 if e.Level == 0 {
    1732            2 :                                         stats.ApproxIngestedIntoL0Bytes += e.Meta.Size
    1733            2 :                                 }
    1734            2 :                                 if metaFlushableOverlaps[e.Meta.TableNum] {
    1735            2 :                                         stats.MemtableOverlappingFiles++
    1736            2 :                                 }
    1737              :                         }
    1738            2 :                 } else if asFlushable {
    1739            2 :                         // NB: If asFlushable == true, there are no shared sstables.
    1740            2 :                         info.Tables = make([]struct {
    1741            2 :                                 TableInfo
    1742            2 :                                 Level int
    1743            2 :                         }, len(loadResult.local))
    1744            2 :                         for i, f := range loadResult.local {
    1745            2 :                                 info.Tables[i].Level = -1
    1746            2 :                                 info.Tables[i].TableInfo = f.TableInfo()
    1747            2 :                                 stats.Bytes += f.Size
    1748            2 :                                 // We don't have exact stats on which files will be ingested into
    1749            2 :                                 // L0, because actual ingestion into the LSM has been deferred until
    1750            2 :                                 // flush time. Instead, we infer based on memtable overlap.
    1751            2 :                                 //
    1752            2 :                                 // TODO(jackson): If we optimistically compute data overlap (#2112)
    1753            2 :                                 // before entering the commit pipeline, we can use that overlap to
    1754            2 :                                 // improve our approximation by incorporating overlap with L0, not
    1755            2 :                                 // just memtables.
    1756            2 :                                 if metaFlushableOverlaps[f.TableNum] {
    1757            2 :                                         stats.ApproxIngestedIntoL0Bytes += f.Size
    1758            2 :                                         stats.MemtableOverlappingFiles++
    1759            2 :                                 }
    1760              :                         }
    1761              :                 }
    1762            2 :                 d.opts.EventListener.TableIngested(info)
    1763              :         }
    1764              : 
    1765            2 :         return stats, err
    1766              : }
    1767              : 
    1768              : type ingestSplitFile struct {
    1769              :         // ingestFile is the file being ingested.
    1770              :         ingestFile *manifest.TableMetadata
    1771              :         // splitFile is the file that needs to be split to allow ingestFile to slot
    1772              :         // into `level` level.
    1773              :         splitFile *manifest.TableMetadata
    1774              :         // The level where ingestFile will go (and where splitFile already is).
    1775              :         level int
    1776              : }
    1777              : 
    1778              : // ingestSplit splits files specified in `files` and updates ve in-place to
    1779              : // account for existing files getting split into two virtual sstables. The map
    1780              : // `replacedFiles` contains an in-progress map of all files that have been
    1781              : // replaced with new virtual sstables in this version edit so far, which is also
    1782              : // updated in-place.
    1783              : //
    1784              : // d.mu as well as the manifest lock must be held when calling this method.
    1785              : func (d *DB) ingestSplit(
    1786              :         ctx context.Context,
    1787              :         ve *manifest.VersionEdit,
    1788              :         updateMetrics func(*manifest.TableMetadata, int, []manifest.NewTableEntry),
    1789              :         files []ingestSplitFile,
    1790              :         replacedTables map[base.TableNum][]manifest.NewTableEntry,
    1791            2 : ) error {
    1792            2 :         for _, s := range files {
    1793            2 :                 ingestFileBounds := s.ingestFile.UserKeyBounds()
    1794            2 :                 // replacedFiles can be thought of as a tree, where we start iterating with
    1795            2 :                 // s.splitFile and run its fileNum through replacedFiles, then find which of
    1796            2 :                 // the replaced files overlaps with s.ingestFile, which becomes the new
    1797            2 :                 // splitFile, then we check splitFile's replacements in replacedFiles again
    1798            2 :                 // for overlap with s.ingestFile, and so on until we either can't find the
    1799            2 :                 // current splitFile in replacedFiles (i.e. that's the file that now needs to
    1800            2 :                 // be split), or we don't find a file that overlaps with s.ingestFile, which
    1801            2 :                 // means a prior ingest split already produced enough room for s.ingestFile
    1802            2 :                 // to go into this level without necessitating another ingest split.
    1803            2 :                 splitFile := s.splitFile
    1804            2 :                 for splitFile != nil {
    1805            2 :                         replaced, ok := replacedTables[splitFile.TableNum]
    1806            2 :                         if !ok {
    1807            2 :                                 break
    1808              :                         }
    1809            2 :                         updatedSplitFile := false
    1810            2 :                         for i := range replaced {
    1811            2 :                                 if replaced[i].Meta.Overlaps(d.cmp, &ingestFileBounds) {
    1812            2 :                                         if updatedSplitFile {
    1813            0 :                                                 // This should never happen because the earlier ingestTargetLevel
    1814            0 :                                                 // function only finds split file candidates that are guaranteed to
    1815            0 :                                                 // have no data overlap, only boundary overlap. See the comments
    1816            0 :                                                 // in that method to see the definitions of data vs boundary
    1817            0 :                                                 // overlap. That, plus the fact that files in `replaced` are
    1818            0 :                                                 // guaranteed to have file bounds that are tight on user keys
    1819            0 :                                                 // (as that's what `d.excise` produces), means that the only case
    1820            0 :                                                 // where we overlap with two or more files in `replaced` is if we
    1821            0 :                                                 // actually had data overlap all along, or if the ingestion files
    1822            0 :                                                 // were overlapping, either of which is an invariant violation.
    1823            0 :                                                 panic("updated with two files in ingestSplit")
    1824              :                                         }
    1825            2 :                                         splitFile = replaced[i].Meta
    1826            2 :                                         updatedSplitFile = true
    1827              :                                 }
    1828              :                         }
    1829            2 :                         if !updatedSplitFile {
    1830            2 :                                 // None of the replaced files overlapped with the file being ingested.
    1831            2 :                                 // This can happen if we've already excised a span overlapping with
    1832            2 :                                 // this file, or if we have consecutive ingested files that can slide
    1833            2 :                                 // within the same gap between keys in an existing file. For instance,
    1834            2 :                                 // if an existing file has keys a and g and we're ingesting b-c, d-e,
    1835            2 :                                 // the first loop iteration will split the existing file into one that
    1836            2 :                                 // ends in a and another that starts at g, and the second iteration will
    1837            2 :                                 // fall into this case and require no splitting.
    1838            2 :                                 //
    1839            2 :                                 // No splitting necessary.
    1840            2 :                                 splitFile = nil
    1841            2 :                         }
    1842              :                 }
    1843            2 :                 if splitFile == nil {
    1844            2 :                         continue
    1845              :                 }
    1846              :                 // NB: excise operates on [start, end). We're splitting at [start, end]
    1847              :                 // (assuming !s.ingestFile.Largest.IsExclusiveSentinel()). The conflation
    1848              :                 // of exclusive vs inclusive end bounds should not make a difference here
    1849              :                 // as we're guaranteed to not have any data overlap between splitFile and
    1850              :                 // s.ingestFile. d.excise will return an error if we pass an inclusive user
    1851              :                 // key bound _and_ we end up seeing data overlap at the end key.
    1852            2 :                 exciseBounds := base.UserKeyBoundsFromInternal(s.ingestFile.Smallest(), s.ingestFile.Largest())
    1853            2 :                 leftTable, rightTable, err := d.exciseTable(ctx, exciseBounds, splitFile, s.level, tightExciseBounds)
    1854            2 :                 if err != nil {
    1855            0 :                         return err
    1856            0 :                 }
    1857            2 :                 added := applyExciseToVersionEdit(ve, splitFile, leftTable, rightTable, s.level)
    1858            2 :                 replacedTables[splitFile.TableNum] = added
    1859            2 :                 for i := range added {
    1860            2 :                         addedBounds := added[i].Meta.UserKeyBounds()
    1861            2 :                         if s.ingestFile.Overlaps(d.cmp, &addedBounds) {
    1862            0 :                                 panic("ingest-time split produced a file that overlaps with ingested file")
    1863              :                         }
    1864              :                 }
    1865            2 :                 updateMetrics(splitFile, s.level, added)
    1866              :         }
    1867              :         // Flatten the version edit by removing any entries from ve.NewFiles that
    1868              :         // are also in ve.DeletedFiles.
    1869            2 :         newNewFiles := ve.NewTables[:0]
    1870            2 :         for i := range ve.NewTables {
    1871            2 :                 fn := ve.NewTables[i].Meta.TableNum
    1872            2 :                 deEntry := manifest.DeletedTableEntry{Level: ve.NewTables[i].Level, FileNum: fn}
    1873            2 :                 if _, ok := ve.DeletedTables[deEntry]; ok {
    1874            2 :                         delete(ve.DeletedTables, deEntry)
    1875            2 :                 } else {
    1876            2 :                         newNewFiles = append(newNewFiles, ve.NewTables[i])
    1877            2 :                 }
    1878              :         }
    1879            2 :         ve.NewTables = newNewFiles
    1880            2 :         return nil
    1881              : }
    1882              : 
    1883              : func (d *DB) ingestApply(
    1884              :         ctx context.Context,
    1885              :         jobID JobID,
    1886              :         lr ingestLoadResult,
    1887              :         mut *memTable,
    1888              :         exciseSpan KeyRange,
    1889              :         exciseBoundsPolicy exciseBoundsPolicy,
    1890              :         exciseSeqNum base.SeqNum,
    1891            2 : ) (*manifest.VersionEdit, error) {
    1892            2 :         d.mu.Lock()
    1893            2 :         defer d.mu.Unlock()
    1894            2 : 
    1895            2 :         ve := &manifest.VersionEdit{
    1896            2 :                 NewTables: make([]manifest.NewTableEntry, lr.fileCount()),
    1897            2 :         }
    1898            2 :         if exciseSpan.Valid() || (d.opts.Experimental.IngestSplit != nil && d.opts.Experimental.IngestSplit()) {
    1899            2 :                 ve.DeletedTables = map[manifest.DeletedTableEntry]*manifest.TableMetadata{}
    1900            2 :         }
    1901            2 :         var metrics levelMetricsDelta
    1902            2 : 
    1903            2 :         // Determine the target level inside UpdateVersionLocked. This prevents two
    1904            2 :         // concurrent ingestion jobs from using the same version to determine the
    1905            2 :         // target level, and also provides serialization with concurrent compaction
    1906            2 :         // and flush jobs.
    1907            2 :         err := d.mu.versions.UpdateVersionLocked(func() (versionUpdate, error) {
    1908            2 :                 if mut != nil {
    1909            2 :                         // Unref the mutable memtable to allows its flush to proceed. Now that we've
    1910            2 :                         // acquired the manifest lock, we can be certain that if the mutable
    1911            2 :                         // memtable has received more recent conflicting writes, the flush won't
    1912            2 :                         // beat us to applying to the manifest resulting in sequence number
    1913            2 :                         // inversion. Even though we call maybeScheduleFlush right now, this flush
    1914            2 :                         // will apply after our ingestion.
    1915            2 :                         if mut.writerUnref() {
    1916            2 :                                 d.maybeScheduleFlush()
    1917            2 :                         }
    1918              :                 }
    1919              : 
    1920            2 :                 current := d.mu.versions.currentVersion()
    1921            2 :                 overlapChecker := &overlapChecker{
    1922            2 :                         comparer: d.opts.Comparer,
    1923            2 :                         newIters: d.newIters,
    1924            2 :                         opts: IterOptions{
    1925            2 :                                 logger:   d.opts.Logger,
    1926            2 :                                 Category: categoryIngest,
    1927            2 :                         },
    1928            2 :                         v: current,
    1929            2 :                 }
    1930            2 :                 shouldIngestSplit := d.opts.Experimental.IngestSplit != nil &&
    1931            2 :                         d.opts.Experimental.IngestSplit() && d.FormatMajorVersion() >= FormatVirtualSSTables
    1932            2 :                 baseLevel := d.mu.versions.picker.getBaseLevel()
    1933            2 :                 // filesToSplit is a list where each element is a pair consisting of a file
    1934            2 :                 // being ingested and a file being split to make room for an ingestion into
    1935            2 :                 // that level. Each ingested file will appear at most once in this list. It
    1936            2 :                 // is possible for split files to appear twice in this list.
    1937            2 :                 filesToSplit := make([]ingestSplitFile, 0)
    1938            2 :                 checkCompactions := false
    1939            2 :                 for i := 0; i < lr.fileCount(); i++ {
    1940            2 :                         // Determine the lowest level in the LSM for which the sstable doesn't
    1941            2 :                         // overlap any existing files in the level.
    1942            2 :                         var m *manifest.TableMetadata
    1943            2 :                         specifiedLevel := -1
    1944            2 :                         isShared := false
    1945            2 :                         isExternal := false
    1946            2 :                         if i < len(lr.local) {
    1947            2 :                                 // local file.
    1948            2 :                                 m = lr.local[i].TableMetadata
    1949            2 :                         } else if (i - len(lr.local)) < len(lr.shared) {
    1950            2 :                                 // shared file.
    1951            2 :                                 isShared = true
    1952            2 :                                 sharedIdx := i - len(lr.local)
    1953            2 :                                 m = lr.shared[sharedIdx].TableMetadata
    1954            2 :                                 specifiedLevel = int(lr.shared[sharedIdx].shared.Level)
    1955            2 :                         } else {
    1956            2 :                                 // external file.
    1957            2 :                                 isExternal = true
    1958            2 :                                 externalIdx := i - (len(lr.local) + len(lr.shared))
    1959            2 :                                 m = lr.external[externalIdx].TableMetadata
    1960            2 :                                 if lr.externalFilesHaveLevel {
    1961            1 :                                         specifiedLevel = int(lr.external[externalIdx].external.Level)
    1962            1 :                                 }
    1963              :                         }
    1964              : 
    1965              :                         // Add to CreatedBackingTables if this is a new backing.
    1966              :                         //
    1967              :                         // Shared files always have a new backing. External files have new backings
    1968              :                         // iff the backing disk file num and the file num match (see ingestAttachRemote).
    1969            2 :                         if isShared || (isExternal && m.TableBacking.DiskFileNum == base.DiskFileNum(m.TableNum)) {
    1970            2 :                                 ve.CreatedBackingTables = append(ve.CreatedBackingTables, m.TableBacking)
    1971            2 :                         }
    1972              : 
    1973            2 :                         f := &ve.NewTables[i]
    1974            2 :                         var err error
    1975            2 :                         if specifiedLevel != -1 {
    1976            2 :                                 f.Level = specifiedLevel
    1977            2 :                         } else {
    1978            2 :                                 var splitTable *manifest.TableMetadata
    1979            2 :                                 if exciseSpan.Valid() && exciseSpan.Contains(d.cmp, m.Smallest()) && exciseSpan.Contains(d.cmp, m.Largest()) {
    1980            2 :                                         // This file fits perfectly within the excise span. We can slot it at
    1981            2 :                                         // L6, or sharedLevelsStart - 1 if we have shared files.
    1982            2 :                                         if len(lr.shared) > 0 || lr.externalFilesHaveLevel {
    1983            2 :                                                 f.Level = sharedLevelsStart - 1
    1984            2 :                                                 if baseLevel > f.Level {
    1985            2 :                                                         f.Level = 0
    1986            2 :                                                 }
    1987            2 :                                         } else {
    1988            2 :                                                 f.Level = 6
    1989            2 :                                         }
    1990            2 :                                 } else {
    1991            2 :                                         // We check overlap against the LSM without holding DB.mu. Note that we
    1992            2 :                                         // are still holding the log lock, so the version cannot change.
    1993            2 :                                         // TODO(radu): perform this check optimistically outside of the log lock.
    1994            2 :                                         var lsmOverlap overlap.WithLSM
    1995            2 :                                         lsmOverlap, err = func() (overlap.WithLSM, error) {
    1996            2 :                                                 d.mu.Unlock()
    1997            2 :                                                 defer d.mu.Lock()
    1998            2 :                                                 return overlapChecker.DetermineLSMOverlap(ctx, m.UserKeyBounds())
    1999            2 :                                         }()
    2000            2 :                                         if err == nil {
    2001            2 :                                                 f.Level, splitTable, err = ingestTargetLevel(
    2002            2 :                                                         ctx, d.cmp, lsmOverlap, baseLevel, d.mu.compact.inProgress, m, shouldIngestSplit,
    2003            2 :                                                 )
    2004            2 :                                         }
    2005              :                                 }
    2006              : 
    2007            2 :                                 if splitTable != nil {
    2008            2 :                                         if invariants.Enabled {
    2009            2 :                                                 if lf := current.Levels[f.Level].Find(d.cmp, splitTable); lf.Empty() {
    2010            0 :                                                         panic("splitFile returned is not in level it should be")
    2011              :                                                 }
    2012              :                                         }
    2013              :                                         // We take advantage of the fact that we won't drop the db mutex
    2014              :                                         // between now and the call to UpdateVersionLocked. So, no files should
    2015              :                                         // get added to a new in-progress compaction at this point. We can
    2016              :                                         // avoid having to iterate on in-progress compactions to cancel them
    2017              :                                         // if none of the files being split have a compacting state.
    2018            2 :                                         if splitTable.IsCompacting() {
    2019            1 :                                                 checkCompactions = true
    2020            1 :                                         }
    2021            2 :                                         filesToSplit = append(filesToSplit, ingestSplitFile{ingestFile: m, splitFile: splitTable, level: f.Level})
    2022              :                                 }
    2023              :                         }
    2024            2 :                         if err != nil {
    2025            0 :                                 return versionUpdate{}, err
    2026            0 :                         }
    2027            2 :                         if isShared && f.Level < sharedLevelsStart {
    2028            0 :                                 panic(fmt.Sprintf("cannot slot a shared file higher than the highest shared level: %d < %d",
    2029            0 :                                         f.Level, sharedLevelsStart))
    2030              :                         }
    2031            2 :                         f.Meta = m
    2032            2 :                         levelMetrics := metrics[f.Level]
    2033            2 :                         if levelMetrics == nil {
    2034            2 :                                 levelMetrics = &LevelMetrics{}
    2035            2 :                                 metrics[f.Level] = levelMetrics
    2036            2 :                         }
    2037            2 :                         levelMetrics.TablesCount++
    2038            2 :                         levelMetrics.TablesSize += int64(m.Size)
    2039            2 :                         levelMetrics.EstimatedReferencesSize += m.EstimatedReferenceSize()
    2040            2 :                         levelMetrics.TableBytesIngested += m.Size
    2041            2 :                         levelMetrics.TablesIngested++
    2042              :                 }
    2043              :                 // replacedTables maps files excised due to exciseSpan (or splitFiles returned
    2044              :                 // by ingestTargetLevel), to files that were created to replace it. This map
    2045              :                 // is used to resolve references to split files in filesToSplit, as it is
    2046              :                 // possible for a file that we want to split to no longer exist or have a
    2047              :                 // newer fileMetadata due to a split induced by another ingestion file, or an
    2048              :                 // excise.
    2049            2 :                 replacedTables := make(map[base.TableNum][]manifest.NewTableEntry)
    2050            2 :                 updateLevelMetricsOnExcise := func(m *manifest.TableMetadata, level int, added []manifest.NewTableEntry) {
    2051            2 :                         levelMetrics := metrics[level]
    2052            2 :                         if levelMetrics == nil {
    2053            2 :                                 levelMetrics = &LevelMetrics{}
    2054            2 :                                 metrics[level] = levelMetrics
    2055            2 :                         }
    2056            2 :                         levelMetrics.TablesCount--
    2057            2 :                         levelMetrics.TablesSize -= int64(m.Size)
    2058            2 :                         levelMetrics.EstimatedReferencesSize -= m.EstimatedReferenceSize()
    2059            2 :                         for i := range added {
    2060            2 :                                 levelMetrics.TablesCount++
    2061            2 :                                 levelMetrics.TablesSize += int64(added[i].Meta.Size)
    2062            2 :                                 levelMetrics.EstimatedReferencesSize += added[i].Meta.EstimatedReferenceSize()
    2063            2 :                         }
    2064              :                 }
    2065            2 :                 if exciseSpan.Valid() {
    2066            2 :                         exciseBounds := exciseSpan.UserKeyBounds()
    2067            2 :                         // Iterate through all levels and find files that intersect with exciseSpan.
    2068            2 :                         //
    2069            2 :                         // TODO(bilal): We could drop the DB mutex here as we don't need it for
    2070            2 :                         // excises; we only need to hold the version lock which we already are
    2071            2 :                         // holding. However releasing the DB mutex could mess with the
    2072            2 :                         // ingestTargetLevel calculation that happened above, as it assumed that it
    2073            2 :                         // had a complete view of in-progress compactions that wouldn't change
    2074            2 :                         // until UpdateVersionLocked is called. If we were to drop the mutex now,
    2075            2 :                         // we could schedule another in-progress compaction that would go into the
    2076            2 :                         // chosen target level and lead to file overlap within level (which would
    2077            2 :                         // panic in UpdateVersionLocked). We should drop the db mutex here, do the
    2078            2 :                         // excise, then re-grab the DB mutex and rerun just the in-progress
    2079            2 :                         // compaction check to see if any new compactions are conflicting with our
    2080            2 :                         // chosen target levels for files, and if they are, we should signal those
    2081            2 :                         // compactions to error out.
    2082            2 :                         for layer, ls := range current.AllLevelsAndSublevels() {
    2083            2 :                                 for m := range ls.Overlaps(d.cmp, exciseSpan.UserKeyBounds()).All() {
    2084            2 :                                         leftTable, rightTable, err := d.exciseTable(ctx, exciseBounds, m, layer.Level(), exciseBoundsPolicy)
    2085            2 :                                         if err != nil {
    2086            0 :                                                 return versionUpdate{}, err
    2087            0 :                                         }
    2088            2 :                                         newFiles := applyExciseToVersionEdit(ve, m, leftTable, rightTable, layer.Level())
    2089            2 :                                         replacedTables[m.TableNum] = newFiles
    2090            2 :                                         updateLevelMetricsOnExcise(m, layer.Level(), newFiles)
    2091              :                                 }
    2092              :                         }
    2093              :                 }
    2094            2 :                 if len(filesToSplit) > 0 {
    2095            2 :                         // For the same reasons as the above call to excise, we hold the db mutex
    2096            2 :                         // while calling this method.
    2097            2 :                         if err := d.ingestSplit(ctx, ve, updateLevelMetricsOnExcise, filesToSplit, replacedTables); err != nil {
    2098            0 :                                 return versionUpdate{}, err
    2099            0 :                         }
    2100              :                 }
    2101            2 :                 if len(filesToSplit) > 0 || exciseSpan.Valid() {
    2102            2 :                         for c := range d.mu.compact.inProgress {
    2103            2 :                                 if c.versionEditApplied {
    2104            1 :                                         continue
    2105              :                                 }
    2106              :                                 // Check if this compaction overlaps with the excise span. Note that just
    2107              :                                 // checking if the inputs individually overlap with the excise span
    2108              :                                 // isn't sufficient; for instance, a compaction could have [a,b] and [e,f]
    2109              :                                 // as inputs and write it all out as [a,b,e,f] in one sstable. If we're
    2110              :                                 // doing a [c,d) excise at the same time as this compaction, we will have
    2111              :                                 // to error out the whole compaction as we can't guarantee it hasn't/won't
    2112              :                                 // write a file overlapping with the excise span.
    2113            2 :                                 if exciseSpan.OverlapsInternalKeyRange(d.cmp, c.smallest, c.largest) {
    2114            2 :                                         c.cancel.Store(true)
    2115            2 :                                 }
    2116              :                                 // Check if this compaction's inputs have been replaced due to an
    2117              :                                 // ingest-time split. In that case, cancel the compaction as a newly picked
    2118              :                                 // compaction would need to include any new files that slid in between
    2119              :                                 // previously-existing files. Note that we cancel any compaction that has a
    2120              :                                 // file that was ingest-split as an input, even if it started before this
    2121              :                                 // ingestion.
    2122            2 :                                 if checkCompactions {
    2123            1 :                                         for i := range c.inputs {
    2124            1 :                                                 for f := range c.inputs[i].files.All() {
    2125            1 :                                                         if _, ok := replacedTables[f.TableNum]; ok {
    2126            1 :                                                                 c.cancel.Store(true)
    2127            1 :                                                                 break
    2128              :                                                         }
    2129              :                                                 }
    2130              :                                         }
    2131              :                                 }
    2132              :                         }
    2133              :                 }
    2134              : 
    2135            2 :                 return versionUpdate{
    2136            2 :                         VE:                      ve,
    2137            2 :                         JobID:                   jobID,
    2138            2 :                         Metrics:                 metrics,
    2139            2 :                         InProgressCompactionsFn: func() []compactionInfo { return d.getInProgressCompactionInfoLocked(nil) },
    2140              :                 }, nil
    2141              :         })
    2142            2 :         if err != nil {
    2143            1 :                 return nil, err
    2144            1 :         }
    2145              : 
    2146              :         // Check for any EventuallyFileOnlySnapshots that could be watching for
    2147              :         // an excise on this span. There should be none as the
    2148              :         // computePossibleOverlaps steps should have forced these EFOS to transition
    2149              :         // to file-only snapshots by now. If we see any that conflict with this
    2150              :         // excise, panic.
    2151            2 :         if exciseSpan.Valid() {
    2152            2 :                 for s := d.mu.snapshots.root.next; s != &d.mu.snapshots.root; s = s.next {
    2153            1 :                         // Skip non-EFOS snapshots, and also skip any EFOS that were created
    2154            1 :                         // *after* the excise.
    2155            1 :                         if s.efos == nil || base.Visible(exciseSeqNum, s.efos.seqNum, base.SeqNumMax) {
    2156            0 :                                 continue
    2157              :                         }
    2158            1 :                         efos := s.efos
    2159            1 :                         // TODO(bilal): We can make this faster by taking advantage of the sorted
    2160            1 :                         // nature of protectedRanges to do a sort.Search, or even maintaining a
    2161            1 :                         // global list of all protected ranges instead of having to peer into every
    2162            1 :                         // snapshot.
    2163            1 :                         for i := range efos.protectedRanges {
    2164            1 :                                 if efos.protectedRanges[i].OverlapsKeyRange(d.cmp, exciseSpan) {
    2165            0 :                                         panic("unexpected excise of an EventuallyFileOnlySnapshot's bounds")
    2166              :                                 }
    2167              :                         }
    2168              :                 }
    2169              :         }
    2170              : 
    2171            2 :         d.mu.versions.metrics.Ingest.Count++
    2172            2 : 
    2173            2 :         d.updateReadStateLocked(d.opts.DebugCheck)
    2174            2 :         // updateReadStateLocked could have generated obsolete tables, schedule a
    2175            2 :         // cleanup job if necessary.
    2176            2 :         d.deleteObsoleteFiles(jobID)
    2177            2 :         d.updateTableStatsLocked(ve.NewTables)
    2178            2 :         // The ingestion may have pushed a level over the threshold for compaction,
    2179            2 :         // so check to see if one is necessary and schedule it.
    2180            2 :         d.maybeScheduleCompaction()
    2181            2 :         var toValidate []manifest.NewTableEntry
    2182            2 :         dedup := make(map[base.DiskFileNum]struct{})
    2183            2 :         for _, entry := range ve.NewTables {
    2184            2 :                 if _, ok := dedup[entry.Meta.TableBacking.DiskFileNum]; !ok {
    2185            2 :                         toValidate = append(toValidate, entry)
    2186            2 :                         dedup[entry.Meta.TableBacking.DiskFileNum] = struct{}{}
    2187            2 :                 }
    2188              :         }
    2189            2 :         d.maybeValidateSSTablesLocked(toValidate)
    2190            2 :         return ve, nil
    2191              : }
    2192              : 
    2193              : // maybeValidateSSTablesLocked adds the slice of newTableEntrys to the pending
    2194              : // queue of files to be validated, when the feature is enabled.
    2195              : //
    2196              : // Note that if two entries with the same backing file are added twice, then the
    2197              : // block checksums for the backing file will be validated twice.
    2198              : //
    2199              : // DB.mu must be locked when calling.
    2200            2 : func (d *DB) maybeValidateSSTablesLocked(newFiles []manifest.NewTableEntry) {
    2201            2 :         // Only add to the validation queue when the feature is enabled.
    2202            2 :         if !d.opts.Experimental.ValidateOnIngest {
    2203            2 :                 return
    2204            2 :         }
    2205              : 
    2206            2 :         d.mu.tableValidation.pending = append(d.mu.tableValidation.pending, newFiles...)
    2207            2 :         if d.shouldValidateSSTablesLocked() {
    2208            2 :                 go d.validateSSTables()
    2209            2 :         }
    2210              : }
    2211              : 
    2212              : // shouldValidateSSTablesLocked returns true if SSTable validation should run.
    2213              : // DB.mu must be locked when calling.
    2214            2 : func (d *DB) shouldValidateSSTablesLocked() bool {
    2215            2 :         return !d.mu.tableValidation.validating &&
    2216            2 :                 d.closed.Load() == nil &&
    2217            2 :                 d.opts.Experimental.ValidateOnIngest &&
    2218            2 :                 len(d.mu.tableValidation.pending) > 0
    2219            2 : }
    2220              : 
    2221              : // validateSSTables runs a round of validation on the tables in the pending
    2222              : // queue.
    2223            2 : func (d *DB) validateSSTables() {
    2224            2 :         d.mu.Lock()
    2225            2 :         if !d.shouldValidateSSTablesLocked() {
    2226            2 :                 d.mu.Unlock()
    2227            2 :                 return
    2228            2 :         }
    2229              : 
    2230            2 :         pending := d.mu.tableValidation.pending
    2231            2 :         d.mu.tableValidation.pending = nil
    2232            2 :         d.mu.tableValidation.validating = true
    2233            2 :         jobID := d.newJobIDLocked()
    2234            2 :         rs := d.loadReadState()
    2235            2 : 
    2236            2 :         // Drop DB.mu before performing IO.
    2237            2 :         d.mu.Unlock()
    2238            2 : 
    2239            2 :         // Validate all tables in the pending queue. This could lead to a situation
    2240            2 :         // where we are starving IO from other tasks due to having to page through
    2241            2 :         // all the blocks in all the sstables in the queue.
    2242            2 :         // TODO(travers): Add some form of pacing to avoid IO starvation.
    2243            2 : 
    2244            2 :         // If we fail to validate any files due to reasons other than uncovered
    2245            2 :         // corruption, accumulate them and re-queue them for another attempt.
    2246            2 :         var retry []manifest.NewTableEntry
    2247            2 : 
    2248            2 :         for _, f := range pending {
    2249            2 :                 // The file may have been moved or deleted since it was ingested, in
    2250            2 :                 // which case we skip.
    2251            2 :                 if !rs.current.Contains(f.Level, f.Meta) {
    2252            2 :                         // Assume the file was moved to a lower level. It is rare enough
    2253            2 :                         // that a table is moved or deleted between the time it was ingested
    2254            2 :                         // and the time the validation routine runs that the overall cost of
    2255            2 :                         // this inner loop is tolerably low, when amortized over all
    2256            2 :                         // ingested tables.
    2257            2 :                         found := false
    2258            2 :                         for i := f.Level + 1; i < numLevels; i++ {
    2259            2 :                                 if rs.current.Contains(i, f.Meta) {
    2260            2 :                                         found = true
    2261            2 :                                         break
    2262              :                                 }
    2263              :                         }
    2264            2 :                         if !found {
    2265            2 :                                 continue
    2266              :                         }
    2267              :                 }
    2268              : 
    2269              :                 // TOOD(radu): plumb a ReadEnv with a CategoryIngest stats collector through
    2270              :                 // to ValidateBlockChecksums.
    2271            2 :                 err := d.fileCache.withReader(context.TODO(), block.NoReadEnv,
    2272            2 :                         f.Meta, func(r *sstable.Reader, _ sstable.ReadEnv) error {
    2273            2 :                                 return r.ValidateBlockChecksums()
    2274            2 :                         })
    2275              : 
    2276            2 :                 if err != nil {
    2277            1 :                         if IsCorruptionError(err) {
    2278            1 :                                 // TODO(travers): Hook into the corruption reporting pipeline, once
    2279            1 :                                 // available. See pebble#1192.
    2280            1 :                                 d.opts.Logger.Fatalf("pebble: encountered corruption during ingestion: %s", err)
    2281            1 :                         } else {
    2282            1 :                                 // If there was some other, possibly transient, error that
    2283            1 :                                 // caused table validation to fail inform the EventListener and
    2284            1 :                                 // move on. We remember the table so that we can retry it in a
    2285            1 :                                 // subsequent table validation job.
    2286            1 :                                 //
    2287            1 :                                 // TODO(jackson): If the error is not transient, this will retry
    2288            1 :                                 // validation indefinitely. While not great, it's the same
    2289            1 :                                 // behavior as erroring flushes and compactions. We should
    2290            1 :                                 // address this as a part of #270.
    2291            1 :                                 d.opts.EventListener.BackgroundError(err)
    2292            1 :                                 retry = append(retry, f)
    2293            1 :                                 continue
    2294              :                         }
    2295              :                 }
    2296              : 
    2297            2 :                 d.opts.EventListener.TableValidated(TableValidatedInfo{
    2298            2 :                         JobID: int(jobID),
    2299            2 :                         Meta:  f.Meta,
    2300            2 :                 })
    2301              :         }
    2302            2 :         rs.unref()
    2303            2 :         d.mu.Lock()
    2304            2 :         defer d.mu.Unlock()
    2305            2 :         d.mu.tableValidation.pending = append(d.mu.tableValidation.pending, retry...)
    2306            2 :         d.mu.tableValidation.validating = false
    2307            2 :         d.mu.tableValidation.cond.Broadcast()
    2308            2 :         if d.shouldValidateSSTablesLocked() {
    2309            2 :                 go d.validateSSTables()
    2310            2 :         }
    2311              : }
        

Generated by: LCOV version 2.0-1