LCOV - code coverage report
Current view: top level - pebble - external_iterator.go (source / functions) Coverage Total Hit
Test: 2025-01-21 08:16Z d2b08c85 - meta test only.lcov Lines: 0.0 % 204 0
Test Date: 2025-01-21 08:17:39 Functions: - 0 0

            Line data    Source code
       1              : // Copyright 2022 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              : 
      10              :         "github.com/cockroachdb/errors"
      11              :         "github.com/cockroachdb/pebble/internal/base"
      12              :         "github.com/cockroachdb/pebble/internal/keyspan"
      13              :         "github.com/cockroachdb/pebble/internal/manifest"
      14              :         "github.com/cockroachdb/pebble/sstable"
      15              :         "github.com/cockroachdb/pebble/sstable/block"
      16              : )
      17              : 
      18              : // NewExternalIter takes an input 2d array of sstable files which may overlap
      19              : // across subarrays but not within a subarray (at least as far as points are
      20              : // concerned; range keys are allowed to overlap arbitrarily even within a
      21              : // subarray), and returns an Iterator over the merged contents of the sstables.
      22              : // Input sstables may contain point keys, range keys, range deletions, etc. The
      23              : // input files slice must be sorted in reverse chronological ordering. A key in a
      24              : // file at a lower index subarray will shadow a key with an identical user key
      25              : // contained within a file at a higher index subarray. Each subarray must be
      26              : // sorted in internal key order, where lower index files contain keys that sort
      27              : // left of files with higher indexes.
      28              : //
      29              : // Input sstables must only contain keys with the zero sequence number.
      30              : //
      31              : // Iterators constructed through NewExternalIter do not support all iterator
      32              : // options, including block-property and table filters. NewExternalIter errors
      33              : // if an incompatible option is set.
      34              : func NewExternalIter(
      35              :         o *Options, iterOpts *IterOptions, files [][]sstable.ReadableFile,
      36            0 : ) (it *Iterator, err error) {
      37            0 :         return NewExternalIterWithContext(context.Background(), o, iterOpts, files)
      38            0 : }
      39              : 
      40              : // NewExternalIterWithContext is like NewExternalIter, and additionally
      41              : // accepts a context for tracing.
      42              : func NewExternalIterWithContext(
      43              :         ctx context.Context, o *Options, iterOpts *IterOptions, files [][]sstable.ReadableFile,
      44            0 : ) (it *Iterator, err error) {
      45            0 :         if iterOpts != nil {
      46            0 :                 if err := validateExternalIterOpts(iterOpts); err != nil {
      47            0 :                         return nil, err
      48            0 :                 }
      49              :         }
      50              : 
      51            0 :         var readers [][]*sstable.Reader
      52            0 : 
      53            0 :         seqNumOffset := 0
      54            0 :         for _, levelFiles := range files {
      55            0 :                 seqNumOffset += len(levelFiles)
      56            0 :         }
      57            0 :         for _, levelFiles := range files {
      58            0 :                 seqNumOffset -= len(levelFiles)
      59            0 :                 subReaders, err := openExternalTables(ctx, o, levelFiles, seqNumOffset, o.MakeReaderOptions())
      60            0 :                 readers = append(readers, subReaders)
      61            0 :                 if err != nil {
      62            0 :                         // Close all the opened readers.
      63            0 :                         for i := range readers {
      64            0 :                                 for j := range readers[i] {
      65            0 :                                         _ = readers[i][j].Close()
      66            0 :                                 }
      67              :                         }
      68            0 :                         return nil, err
      69              :                 }
      70              :         }
      71              : 
      72            0 :         buf := iterAllocPool.Get().(*iterAlloc)
      73            0 :         dbi := &buf.dbi
      74            0 :         *dbi = Iterator{
      75            0 :                 ctx:                 ctx,
      76            0 :                 alloc:               buf,
      77            0 :                 merge:               o.Merger.Merge,
      78            0 :                 comparer:            *o.Comparer,
      79            0 :                 readState:           nil,
      80            0 :                 keyBuf:              buf.keyBuf,
      81            0 :                 prefixOrFullSeekKey: buf.prefixOrFullSeekKey,
      82            0 :                 boundsBuf:           buf.boundsBuf,
      83            0 :                 batch:               nil,
      84            0 :                 // Add the readers to the Iterator so that Close closes them, and
      85            0 :                 // SetOptions can re-construct iterators from them.
      86            0 :                 externalReaders: readers,
      87            0 :                 newIters: func(context.Context, *manifest.FileMetadata, *IterOptions,
      88            0 :                         internalIterOpts, iterKinds) (iterSet, error) {
      89            0 :                         // NB: External iterators are currently constructed without any
      90            0 :                         // `levelIters`. newIters should never be called. When we support
      91            0 :                         // organizing multiple non-overlapping files into a single level
      92            0 :                         // (see TODO below), we'll need to adjust this tableNewIters
      93            0 :                         // implementation to open iterators by looking up f in a map
      94            0 :                         // of readers indexed by *fileMetadata.
      95            0 :                         panic("unreachable")
      96              :                 },
      97              :                 seqNum: base.SeqNumMax,
      98              :         }
      99            0 :         if iterOpts != nil {
     100            0 :                 dbi.opts = *iterOpts
     101            0 :                 dbi.processBounds(iterOpts.LowerBound, iterOpts.UpperBound)
     102            0 :         }
     103            0 :         if err := finishInitializingExternal(ctx, dbi); err != nil {
     104            0 :                 dbi.Close()
     105            0 :                 return nil, err
     106            0 :         }
     107            0 :         return dbi, nil
     108              : }
     109              : 
     110            0 : func validateExternalIterOpts(iterOpts *IterOptions) error {
     111            0 :         switch {
     112            0 :         case iterOpts.PointKeyFilters != nil:
     113            0 :                 return errors.Errorf("pebble: external iterator: PointKeyFilters unsupported")
     114            0 :         case iterOpts.RangeKeyFilters != nil:
     115            0 :                 return errors.Errorf("pebble: external iterator: RangeKeyFilters unsupported")
     116            0 :         case iterOpts.OnlyReadGuaranteedDurable:
     117            0 :                 return errors.Errorf("pebble: external iterator: OnlyReadGuaranteedDurable unsupported")
     118            0 :         case iterOpts.UseL6Filters:
     119            0 :                 return errors.Errorf("pebble: external iterator: UseL6Filters unsupported")
     120              :         }
     121            0 :         return nil
     122              : }
     123              : 
     124            0 : func createExternalPointIter(ctx context.Context, it *Iterator) (topLevelIterator, error) {
     125            0 :         // TODO(jackson): In some instances we could generate fewer levels by using
     126            0 :         // L0Sublevels code to organize nonoverlapping files into the same level.
     127            0 :         // This would allow us to use levelIters and keep a smaller set of data and
     128            0 :         // files in-memory. However, it would also require us to identify the bounds
     129            0 :         // of all the files upfront.
     130            0 : 
     131            0 :         if !it.opts.pointKeys() {
     132            0 :                 return emptyIter, nil
     133            0 :         } else if it.pointIter != nil {
     134            0 :                 return it.pointIter, nil
     135            0 :         }
     136            0 :         mlevels := it.alloc.mlevels[:0]
     137            0 : 
     138            0 :         // TODO(jackson): External iterators never provide categorized iterator
     139            0 :         // stats today because they exist outside the context of a *DB. If the
     140            0 :         // sstables being read are on the physical filesystem, we may still want to
     141            0 :         // thread a CategoryStatsCollector through so that we collect their stats.
     142            0 : 
     143            0 :         if len(it.externalReaders) > cap(mlevels) {
     144            0 :                 mlevels = make([]mergingIterLevel, 0, len(it.externalReaders))
     145            0 :         }
     146              :         // We set a synthetic sequence number, with lower levels having higer numbers.
     147            0 :         seqNum := 0
     148            0 :         for _, readers := range it.externalReaders {
     149            0 :                 seqNum += len(readers)
     150            0 :         }
     151            0 :         for _, readers := range it.externalReaders {
     152            0 :                 for _, r := range readers {
     153            0 :                         var (
     154            0 :                                 rangeDelIter keyspan.FragmentIterator
     155            0 :                                 pointIter    internalIterator
     156            0 :                                 err          error
     157            0 :                         )
     158            0 :                         // We could set hideObsoletePoints=true, since we are reading at
     159            0 :                         // InternalKeySeqNumMax, but we don't bother since these sstables should
     160            0 :                         // not have obsolete points (so the performance optimization is
     161            0 :                         // unnecessary), and we don't want to bother constructing a
     162            0 :                         // BlockPropertiesFilterer that includes obsoleteKeyBlockPropertyFilter.
     163            0 :                         transforms := sstable.IterTransforms{SyntheticSeqNum: sstable.SyntheticSeqNum(seqNum)}
     164            0 :                         seqNum--
     165            0 :                         pointIter, err = r.NewPointIter(
     166            0 :                                 ctx, transforms, it.opts.LowerBound, it.opts.UpperBound, nil, /* BlockPropertiesFilterer */
     167            0 :                                 sstable.NeverUseFilterBlock,
     168            0 :                                 block.ReadEnv{Stats: &it.stats.InternalStats, IterStats: nil},
     169            0 :                                 sstable.MakeTrivialReaderProvider(r))
     170            0 :                         if err != nil {
     171            0 :                                 return nil, err
     172            0 :                         }
     173            0 :                         rangeDelIter, err = r.NewRawRangeDelIter(ctx, sstable.FragmentIterTransforms{
     174            0 :                                 SyntheticSeqNum: sstable.SyntheticSeqNum(seqNum),
     175            0 :                         }, block.ReadEnv{Stats: &it.stats.InternalStats, IterStats: nil})
     176            0 :                         if err != nil {
     177            0 :                                 return nil, err
     178            0 :                         }
     179            0 :                         mlevels = append(mlevels, mergingIterLevel{
     180            0 :                                 iter:         pointIter,
     181            0 :                                 rangeDelIter: rangeDelIter,
     182            0 :                         })
     183              :                 }
     184              :         }
     185              : 
     186            0 :         it.alloc.merging.init(&it.opts, &it.stats.InternalStats, it.comparer.Compare, it.comparer.Split, mlevels...)
     187            0 :         it.alloc.merging.snapshot = base.SeqNumMax
     188            0 :         if len(mlevels) <= cap(it.alloc.levelsPositioned) {
     189            0 :                 it.alloc.merging.levelsPositioned = it.alloc.levelsPositioned[:len(mlevels)]
     190            0 :         }
     191            0 :         return &it.alloc.merging, nil
     192              : }
     193              : 
     194            0 : func finishInitializingExternal(ctx context.Context, it *Iterator) error {
     195            0 :         pointIter, err := createExternalPointIter(ctx, it)
     196            0 :         if err != nil {
     197            0 :                 return err
     198            0 :         }
     199            0 :         it.pointIter = pointIter
     200            0 :         it.iter = it.pointIter
     201            0 : 
     202            0 :         if it.opts.rangeKeys() {
     203            0 :                 it.rangeKeyMasking.init(it, &it.comparer)
     204            0 :                 var rangeKeyIters []keyspan.FragmentIterator
     205            0 :                 if it.rangeKey == nil {
     206            0 :                         // We could take advantage of the lack of overlaps in range keys within
     207            0 :                         // each slice in it.externalReaders, and generate keyspanimpl.LevelIters
     208            0 :                         // out of those. However, since range keys are expected to be sparse to
     209            0 :                         // begin with, the performance gain might not be significant enough to
     210            0 :                         // warrant it.
     211            0 :                         //
     212            0 :                         // TODO(bilal): Explore adding a simpleRangeKeyLevelIter that does not
     213            0 :                         // operate on FileMetadatas (similar to simpleLevelIter), and implements
     214            0 :                         // this optimization.
     215            0 :                         // We set a synthetic sequence number, with lower levels having higer numbers.
     216            0 :                         seqNum := 0
     217            0 :                         for _, readers := range it.externalReaders {
     218            0 :                                 seqNum += len(readers)
     219            0 :                         }
     220            0 :                         for _, readers := range it.externalReaders {
     221            0 :                                 for _, r := range readers {
     222            0 :                                         transforms := sstable.FragmentIterTransforms{SyntheticSeqNum: sstable.SyntheticSeqNum(seqNum)}
     223            0 :                                         seqNum--
     224            0 :                                         if rki, err := r.NewRawRangeKeyIter(ctx, transforms, block.ReadEnv{Stats: &it.stats.InternalStats, IterStats: nil}); err != nil {
     225            0 :                                                 return err
     226            0 :                                         } else if rki != nil {
     227            0 :                                                 rangeKeyIters = append(rangeKeyIters, rki)
     228            0 :                                         }
     229              :                                 }
     230              :                         }
     231            0 :                         if len(rangeKeyIters) > 0 {
     232            0 :                                 it.rangeKey = iterRangeKeyStateAllocPool.Get().(*iteratorRangeKeyState)
     233            0 :                                 it.rangeKey.init(it.comparer.Compare, it.comparer.Split, &it.opts)
     234            0 :                                 it.rangeKey.rangeKeyIter = it.rangeKey.iterConfig.Init(
     235            0 :                                         &it.comparer,
     236            0 :                                         base.SeqNumMax,
     237            0 :                                         it.opts.LowerBound, it.opts.UpperBound,
     238            0 :                                         &it.hasPrefix, &it.prefixOrFullSeekKey,
     239            0 :                                         false /* internalKeys */, &it.rangeKey.internal,
     240            0 :                                 )
     241            0 :                                 for i := range rangeKeyIters {
     242            0 :                                         it.rangeKey.iterConfig.AddLevel(rangeKeyIters[i])
     243            0 :                                 }
     244              :                         }
     245              :                 }
     246            0 :                 if it.rangeKey != nil {
     247            0 :                         it.rangeKey.iiter.Init(&it.comparer, it.iter, it.rangeKey.rangeKeyIter,
     248            0 :                                 keyspan.InterleavingIterOpts{
     249            0 :                                         Mask:       &it.rangeKeyMasking,
     250            0 :                                         LowerBound: it.opts.LowerBound,
     251            0 :                                         UpperBound: it.opts.UpperBound,
     252            0 :                                 })
     253            0 :                         it.iter = &it.rangeKey.iiter
     254            0 :                 }
     255              :         }
     256            0 :         return nil
     257              : }
     258              : 
     259              : func openExternalTables(
     260              :         ctx context.Context,
     261              :         o *Options,
     262              :         files []sstable.ReadableFile,
     263              :         seqNumOffset int,
     264              :         readerOpts sstable.ReaderOptions,
     265            0 : ) (readers []*sstable.Reader, err error) {
     266            0 :         readers = make([]*sstable.Reader, 0, len(files))
     267            0 :         for i := range files {
     268            0 :                 readable, err := sstable.NewSimpleReadable(files[i])
     269            0 :                 if err != nil {
     270            0 :                         return readers, err
     271            0 :                 }
     272            0 :                 r, err := sstable.NewReader(ctx, readable, readerOpts)
     273            0 :                 if err != nil {
     274            0 :                         return readers, err
     275            0 :                 }
     276            0 :                 readers = append(readers, r)
     277              :         }
     278            0 :         return readers, err
     279              : }
        

Generated by: LCOV version 2.0-1