0

Upgrade versions of vendored libs

This commit is contained in:
eleksir
2024-08-03 21:26:21 +03:00
parent 9bac4fb257
commit b6f8cf9155
812 changed files with 56387 additions and 31548 deletions

View File

@@ -10,27 +10,30 @@ import (
"github.com/cockroachdb/errors"
"github.com/cockroachdb/pebble/internal/base"
"github.com/cockroachdb/pebble/internal/invariants"
"github.com/cockroachdb/pebble/internal/keyspan"
"github.com/cockroachdb/pebble/internal/manifest"
"github.com/cockroachdb/pebble/objstorage"
"github.com/cockroachdb/pebble/objstorage/remote"
)
const (
// In skip-shared iteration mode, keys in levels sharedLevelsStart and greater
// (i.e. lower in the LSM) are skipped.
sharedLevelsStart = 5
sharedLevelsStart = remote.SharedLevelsStart
)
// ErrInvalidSkipSharedIteration is returned by ScanInternal if it was called
// with a shared file visitor function, and a file in a shareable level (i.e.
// level >= sharedLevelsStart) was found to not be in shared storage according
// to objstorage.Provider.
var ErrInvalidSkipSharedIteration = errors.New("pebble: cannot use skip-shared iteration due to non-shared files in lower levels")
// to objstorage.Provider, or not shareable for another reason such as for
// containing keys newer than the snapshot sequence number.
var ErrInvalidSkipSharedIteration = errors.New("pebble: cannot use skip-shared iteration due to non-shareable files in lower levels")
// SharedSSTMeta represents an sstable on shared storage that can be ingested
// by another pebble instance. This struct must contain all fields that are
// required for a Pebble instance to ingest a foreign sstable on shared storage,
// including constructing any relevant objstorage.Provider / sharedobjcat.Catalog
// including constructing any relevant objstorage.Provider / remoteobjcat.Catalog
// data structures, as well as creating virtual FileMetadatas.
//
// Note that the Pebble instance creating and returning a SharedSSTMeta might
@@ -40,7 +43,7 @@ var ErrInvalidSkipSharedIteration = errors.New("pebble: cannot use skip-shared i
type SharedSSTMeta struct {
// Backing is the shared object underlying this SST. Can be attached to an
// objstorage.Provider.
Backing objstorage.SharedObjectBackingHandle
Backing objstorage.RemoteObjectBackingHandle
// Smallest and Largest internal keys for the overall bounds. The kind and
// SeqNum of these will reflect what is physically present on the source Pebble
@@ -87,6 +90,12 @@ func (s *SharedSSTMeta) cloneFromFileMeta(f *fileMetadata) {
}
}
type sharedByLevel []SharedSSTMeta
func (s sharedByLevel) Len() int { return len(s) }
func (s sharedByLevel) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
func (s sharedByLevel) Less(i, j int) bool { return s[i].Level < s[j].Level }
type pcIterPos int
const (
@@ -95,16 +104,11 @@ const (
)
// pointCollapsingIterator is an internalIterator that collapses point keys and
// returns at most one point internal key for each user key. Merges are merged,
// sets are emitted as-is, and SingleDeletes are collapsed with the next point
// key, however we don't guarantee that we generate and return a SetWithDel if
// a set shadows a del. Point keys deleted by rangedels are also considered shadowed and not
// exposed.
// returns at most one point internal key for each user key. Merges and
// SingleDels are not supported and result in a panic if encountered. Point keys
// deleted by rangedels are considered shadowed and not exposed.
//
// TODO(bilal): Implement the unimplemented internalIterator methods below.
// Currently this iterator only supports the forward iteration case necessary
// for scanInternal, however foreign sstable iterators will also need to use this
// or a simplified version of this.
// Only used in ScanInternal to return at most one internal key per user key.
type pointCollapsingIterator struct {
iter keyspan.InterleavingIter
pos pcIterPos
@@ -112,32 +116,43 @@ type pointCollapsingIterator struct {
merge base.Merge
err error
seqNum uint64
// The current position of `iter`. Could be backed by savedKey (i.e. iterKey ==
// &savedKey) or could be owned by `iter`. findNextEntry and similar methods
// are expected to save the current value of this to savedKey if they're
// iterating away from the current key but still need to retain it. See
// comments in findNextEntry on how this field is used.
// The current position of `iter`. Always owned by the underlying iter.
iterKey *InternalKey
// The last saved key. findNextEntry and similar methods are expected to save
// the current value of iterKey to savedKey if they're iterating away from the
// current key but still need to retain it. See comments in findNextEntry on
// how this field is used.
//
// At the end of a positioning call:
// - if pos == pcIterPosNext, iterKey is pointing to the next user key owned
// by `iter` while savedKey is holding a copy to our current key.
// - If pos == pcIterPosCur, iterKey is pointing to either a savedKey-backed
// copy of the current key, or an `iter`-owned current key in which case
// the value of savedKey is undefined.
iterKey *InternalKey
savedKey InternalKey
// - If pos == pcIterPosCur, iterKey is pointing to an `iter`-owned current
// key, and savedKey is either undefined or pointing to a version of the
// current key owned by this iterator (i.e. backed by savedKeyBuf).
savedKey InternalKey
savedKeyBuf []byte
// Value at the current iterator position, at iterKey.
value base.LazyValue
// Used for Merge keys only.
valueMerger ValueMerger
valueBuf []byte
iterValue base.LazyValue
// If fixedSeqNum is non-zero, all emitted points are verified to have this
// fixed sequence number.
fixedSeqNum uint64
}
func (p *pointCollapsingIterator) Span() *keyspan.Span {
return p.iter.Span()
}
// SeekPrefixGE implements the InternalIterator interface.
func (p *pointCollapsingIterator) SeekPrefixGE(
prefix, key []byte, flags base.SeekGEFlags,
) (*base.InternalKey, base.LazyValue) {
panic("unimplemented")
p.resetKey()
p.iterKey, p.iterValue = p.iter.SeekPrefixGE(prefix, key, flags)
p.pos = pcIterPosCur
if p.iterKey == nil {
return nil, base.LazyValue{}
}
return p.findNextEntry()
}
// SeekGE implements the InternalIterator interface.
@@ -145,7 +160,7 @@ func (p *pointCollapsingIterator) SeekGE(
key []byte, flags base.SeekGEFlags,
) (*base.InternalKey, base.LazyValue) {
p.resetKey()
p.iterKey, p.value = p.iter.SeekGE(key, flags)
p.iterKey, p.iterValue = p.iter.SeekGE(key, flags)
p.pos = pcIterPosCur
if p.iterKey == nil {
return nil, base.LazyValue{}
@@ -161,67 +176,54 @@ func (p *pointCollapsingIterator) SeekLT(
}
func (p *pointCollapsingIterator) resetKey() {
p.savedKey.UserKey = p.savedKey.UserKey[:0]
p.savedKey.UserKey = p.savedKeyBuf[:0]
p.savedKey.Trailer = 0
p.valueMerger = nil
p.valueBuf = p.valueBuf[:0]
p.iterKey = nil
p.pos = pcIterPosCur
}
// findNextEntry is called to return the next key. p.iter must be positioned at
// the start of the first user key we are interested in.
func (p *pointCollapsingIterator) findNextEntry() (*base.InternalKey, base.LazyValue) {
finishAndReturnMerge := func() (*base.InternalKey, base.LazyValue) {
value, closer, err := p.valueMerger.Finish(true /* includesBase */)
if err != nil {
p.err = err
return nil, base.LazyValue{}
}
p.valueBuf = append(p.valueBuf[:0], value...)
if closer != nil {
_ = closer.Close()
}
p.valueMerger = nil
newValue := base.MakeInPlaceValue(value)
return &p.savedKey, newValue
func (p *pointCollapsingIterator) verifySeqNum(key *base.InternalKey) *base.InternalKey {
if !invariants.Enabled {
return key
}
if p.fixedSeqNum == 0 || key == nil || key.Kind() == InternalKeyKindRangeDelete {
return key
}
if key.SeqNum() != p.fixedSeqNum {
panic(fmt.Sprintf("expected foreign point key to have seqnum %d, got %d", p.fixedSeqNum, key.SeqNum()))
}
return key
}
// saveKey sets p.iterKey (if not-nil) to &p.savedKey. We can use this equality
// as a proxy to determine if we're at the first internal key for a user key.
// findNextEntry is called to return the next key. p.iter must be positioned at the
// start of the first user key we are interested in.
func (p *pointCollapsingIterator) findNextEntry() (*base.InternalKey, base.LazyValue) {
p.saveKey()
// Saves a comparison in the fast path
firstIteration := true
for p.iterKey != nil {
// NB: p.savedKey is either the current key (iff p.iterKey == &p.savedKey),
// NB: p.savedKey is either the current key (iff p.iterKey == firstKey),
// or the previous key.
if p.iterKey != &p.savedKey && !p.comparer.Equal(p.iterKey.UserKey, p.savedKey.UserKey) {
if p.valueMerger != nil {
if p.savedKey.Kind() != InternalKeyKindMerge {
panic(fmt.Sprintf("expected key %s to have MERGE kind", p.iterKey))
}
p.pos = pcIterPosNext
return finishAndReturnMerge()
}
if !firstIteration && !p.comparer.Equal(p.iterKey.UserKey, p.savedKey.UserKey) {
p.saveKey()
continue
}
firstIteration = false
if s := p.iter.Span(); s != nil && s.CoversAt(p.seqNum, p.iterKey.SeqNum()) {
// All future keys for this user key must be deleted.
if p.valueMerger != nil {
return finishAndReturnMerge()
} else if p.savedKey.Kind() == InternalKeyKindSingleDelete {
if p.savedKey.Kind() == InternalKeyKindSingleDelete {
panic("cannot process singledel key in point collapsing iterator")
}
// Fast forward to the next user key.
p.saveKey()
p.iterKey, p.value = p.iter.Next()
p.iterKey, p.iterValue = p.iter.Next()
for p.iterKey != nil && p.savedKey.SeqNum() >= p.iterKey.SeqNum() && p.comparer.Equal(p.iterKey.UserKey, p.savedKey.UserKey) {
p.iterKey, p.value = p.iter.Next()
p.iterKey, p.iterValue = p.iter.Next()
}
continue
}
switch p.savedKey.Kind() {
case InternalKeyKindSet, InternalKeyKindDelete, InternalKeyKindSetWithDelete:
p.saveKey()
case InternalKeyKindSet, InternalKeyKindDelete, InternalKeyKindSetWithDelete, InternalKeyKindDeleteSized:
// Note that we return SETs directly, even if they would otherwise get
// compacted into a Del to turn into a SetWithDelete. This is a fast
// path optimization that can break SINGLEDEL determinism. To lead to
@@ -240,70 +242,23 @@ func (p *pointCollapsingIterator) findNextEntry() (*base.InternalKey, base.LazyV
// of blocks and can determine user key changes without doing key saves
// or comparisons.
p.pos = pcIterPosCur
return p.iterKey, p.value
return p.verifySeqNum(p.iterKey), p.iterValue
case InternalKeyKindSingleDelete:
// Panic, as this iterator is not expected to observe single deletes.
panic("cannot process singledel key in point collapsing iterator")
case InternalKeyKindMerge:
if p.valueMerger == nil {
// Set up merger. This is the first Merge key encountered.
value, callerOwned, err := p.value.Value(p.valueBuf[:0])
if err != nil {
p.err = err
return nil, base.LazyValue{}
}
if !callerOwned {
p.valueBuf = append(p.valueBuf[:0], value...)
} else {
p.valueBuf = value
}
p.valueMerger, err = p.merge(p.iterKey.UserKey, p.valueBuf)
if err != nil {
p.err = err
return nil, base.LazyValue{}
}
p.saveKey()
p.iterKey, p.value = p.iter.Next()
continue
}
switch p.iterKey.Kind() {
case InternalKeyKindSet, InternalKeyKindMerge, InternalKeyKindSetWithDelete:
// Merge into key.
value, callerOwned, err := p.value.Value(p.valueBuf[:0])
if err != nil {
p.err = err
return nil, base.LazyValue{}
}
if !callerOwned {
p.valueBuf = append(p.valueBuf[:0], value...)
} else {
p.valueBuf = value
}
if err := p.valueMerger.MergeOlder(value); err != nil {
p.err = err
return nil, base.LazyValue{}
}
}
if p.iterKey.Kind() != InternalKeyKindMerge {
p.savedKey.SetKind(p.iterKey.Kind())
p.pos = pcIterPosCur
return finishAndReturnMerge()
}
p.iterKey, p.value = p.iter.Next()
// Panic, as this iterator is not expected to observe merges.
panic("cannot process merge key in point collapsing iterator")
case InternalKeyKindRangeDelete:
// These are interleaved by the interleaving iterator ahead of all points.
// We should pass them as-is, but also account for any points ahead of
// them.
p.pos = pcIterPosCur
return p.iterKey, p.value
return p.verifySeqNum(p.iterKey), p.iterValue
default:
panic(fmt.Sprintf("unexpected kind: %d", p.iterKey.Kind()))
}
}
if p.valueMerger != nil {
p.pos = pcIterPosNext
return finishAndReturnMerge()
}
p.resetKey()
return nil, base.LazyValue{}
}
@@ -311,7 +266,7 @@ func (p *pointCollapsingIterator) findNextEntry() (*base.InternalKey, base.LazyV
// First implements the InternalIterator interface.
func (p *pointCollapsingIterator) First() (*base.InternalKey, base.LazyValue) {
p.resetKey()
p.iterKey, p.value = p.iter.First()
p.iterKey, p.iterValue = p.iter.First()
p.pos = pcIterPosCur
if p.iterKey == nil {
return nil, base.LazyValue{}
@@ -326,14 +281,11 @@ func (p *pointCollapsingIterator) Last() (*base.InternalKey, base.LazyValue) {
func (p *pointCollapsingIterator) saveKey() {
if p.iterKey == nil {
p.savedKey = InternalKey{UserKey: p.savedKey.UserKey[:0]}
return
} else if p.iterKey == &p.savedKey {
// Do nothing.
p.savedKey = InternalKey{UserKey: p.savedKeyBuf[:0]}
return
}
p.savedKey.CopyFrom(*p.iterKey)
p.iterKey = &p.savedKey
p.savedKeyBuf = append(p.savedKeyBuf[:0], p.iterKey.UserKey...)
p.savedKey = InternalKey{UserKey: p.savedKeyBuf, Trailer: p.iterKey.Trailer}
}
// Next implements the InternalIterator interface.
@@ -345,17 +297,16 @@ func (p *pointCollapsingIterator) Next() (*base.InternalKey, base.LazyValue) {
// Step over the interleaved range delete and process the very next
// internal key, even if it's at the same user key. This is because a
// point for that user key has not been returned yet.
p.iterKey, p.value = p.iter.Next()
p.iterKey, p.iterValue = p.iter.Next()
break
}
// Fast forward to the next user key. p.iterKey stayed at p.savedKey for
// this loop.
// Fast forward to the next user key.
key, val := p.iter.Next()
// p.iterKey.SeqNum() >= key.SeqNum() is an optimization that allows us to
// use p.iterKey.SeqNum() < key.SeqNum() as a sign that the user key has
// changed, without needing to do the full key comparison.
for p.iterKey != nil && key != nil && p.iterKey.SeqNum() >= key.SeqNum() &&
p.comparer.Equal(p.iterKey.UserKey, key.UserKey) {
for key != nil && p.savedKey.SeqNum() >= key.SeqNum() &&
p.comparer.Equal(p.savedKey.UserKey, key.UserKey) {
key, val = p.iter.Next()
}
if key == nil {
@@ -363,7 +314,7 @@ func (p *pointCollapsingIterator) Next() (*base.InternalKey, base.LazyValue) {
p.resetKey()
return nil, base.LazyValue{}
}
p.iterKey, p.value = key, val
p.iterKey, p.iterValue = key, val
case pcIterPosNext:
p.pos = pcIterPosCur
}
@@ -410,6 +361,32 @@ func (p *pointCollapsingIterator) String() string {
var _ internalIterator = &pointCollapsingIterator{}
// IteratorLevelKind is used to denote whether the current ScanInternal iterator
// is unknown, belongs to a flushable, or belongs to an LSM level type.
type IteratorLevelKind int8
const (
// IteratorLevelUnknown indicates an unknown LSM level.
IteratorLevelUnknown IteratorLevelKind = iota
// IteratorLevelLSM indicates an LSM level.
IteratorLevelLSM
// IteratorLevelFlushable indicates a flushable (i.e. memtable).
IteratorLevelFlushable
)
// IteratorLevel is used with scanInternalIterator to surface additional iterator-specific info where possible.
// Note: this is struct is only provided for point keys.
type IteratorLevel struct {
Kind IteratorLevelKind
// FlushableIndex indicates the position within the flushable queue of this level.
// Only valid if kind == IteratorLevelFlushable.
FlushableIndex int
// The level within the LSM. Only valid if Kind == IteratorLevelLSM.
Level int
// Sublevel is only valid if Kind == IteratorLevelLSM and Level == 0.
Sublevel int
}
// scanInternalIterator is an iterator that returns all internal keys, including
// tombstones. For instance, an InternalKeyKindDelete would be returned as an
// InternalKeyKindDelete instead of the iterator skipping over to the next key.
@@ -423,19 +400,23 @@ var _ internalIterator = &pointCollapsingIterator{}
// *must* return the range delete as well as the range key unset/delete that did
// the shadowing.
type scanInternalIterator struct {
db *DB
opts scanInternalOptions
comparer *base.Comparer
merge Merge
iter internalIterator
readState *readState
version *version
rangeKey *iteratorRangeKeyState
pointKeyIter pointCollapsingIterator
pointKeyIter internalIterator
iterKey *InternalKey
iterValue LazyValue
alloc *iterAlloc
newIters tableNewIters
newIterRangeKey keyspan.TableNewSpanIter
seqNum uint64
iterLevels []IteratorLevel
mergingIter *mergingIter
// boundsBuf holds two buffers used to store the lower and upper bounds.
// Whenever the InternalIterator's bounds change, the new bounds are copied
@@ -464,7 +445,7 @@ func (d *DB) truncateSharedFile(
sst = &SharedSSTMeta{}
sst.cloneFromFileMeta(file)
sst.Level = uint8(level)
sst.Backing, err = d.objProvider.SharedObjectBacking(&objMeta)
sst.Backing, err = d.objProvider.RemoteObjectBacking(&objMeta)
if err != nil {
return nil, false, err
}
@@ -477,10 +458,6 @@ func (d *DB) truncateSharedFile(
// We will need to truncate file bounds in at least one direction. Open all
// relevant iterators.
//
// TODO(bilal): Once virtual sstables go in, verify that the constraining of
// bounds to virtual sstable bounds happens below this method, so we aren't
// unintentionally exposing keys we shouldn't be exposing.
iter, rangeDelIter, err := d.newIters(ctx, file, &IterOptions{
LowerBound: lower,
UpperBound: upper,
@@ -493,18 +470,18 @@ func (d *DB) truncateSharedFile(
if rangeDelIter != nil {
rangeDelIter = keyspan.Truncate(
cmp, rangeDelIter, lower, upper, nil, nil,
false, /* panicOnPartialOverlap */
false, /* panicOnUpperTruncate */
)
defer rangeDelIter.Close()
}
rangeKeyIter, err := d.tableNewRangeKeyIter(file, nil /* spanIterOptions */)
rangeKeyIter, err := d.tableNewRangeKeyIter(file, keyspan.SpanIterOptions{})
if err != nil {
return nil, false, err
}
if rangeKeyIter != nil {
rangeKeyIter = keyspan.Truncate(
cmp, rangeKeyIter, lower, upper, nil, nil,
false, /* panicOnPartialOverlap */
false, /* panicOnUpperTruncate */
)
defer rangeKeyIter.Close()
}
@@ -514,6 +491,7 @@ func (d *DB) truncateSharedFile(
sst.SmallestPointKey.UserKey = sst.SmallestPointKey.UserKey[:0]
sst.SmallestPointKey.Trailer = 0
key, _ := iter.SeekGE(lower, base.SeekGEFlagsNone)
foundPointKey := key != nil
if key != nil {
sst.SmallestPointKey.CopyFrom(*key)
}
@@ -521,14 +499,24 @@ func (d *DB) truncateSharedFile(
span := rangeDelIter.SeekGE(lower)
if span != nil && (len(sst.SmallestPointKey.UserKey) == 0 || base.InternalCompare(cmp, span.SmallestKey(), sst.SmallestPointKey) < 0) {
sst.SmallestPointKey.CopyFrom(span.SmallestKey())
foundPointKey = true
}
}
if !foundPointKey {
// There are no point keys in the span we're interested in.
sst.SmallestPointKey = InternalKey{}
sst.LargestPointKey = InternalKey{}
}
sst.SmallestRangeKey.UserKey = sst.SmallestRangeKey.UserKey[:0]
sst.SmallestRangeKey.Trailer = 0
if rangeKeyIter != nil {
span := rangeKeyIter.SeekGE(lower)
if span != nil {
sst.SmallestRangeKey.CopyFrom(span.SmallestKey())
} else {
// There are no range keys in the span we're interested in.
sst.SmallestRangeKey = InternalKey{}
sst.LargestRangeKey = InternalKey{}
}
}
}
@@ -538,6 +526,7 @@ func (d *DB) truncateSharedFile(
sst.LargestPointKey.UserKey = sst.LargestPointKey.UserKey[:0]
sst.LargestPointKey.Trailer = 0
key, _ := iter.SeekLT(upper, base.SeekLTFlagsNone)
foundPointKey := key != nil
if key != nil {
sst.LargestPointKey.CopyFrom(*key)
}
@@ -545,14 +534,24 @@ func (d *DB) truncateSharedFile(
span := rangeDelIter.SeekLT(upper)
if span != nil && (len(sst.LargestPointKey.UserKey) == 0 || base.InternalCompare(cmp, span.LargestKey(), sst.LargestPointKey) > 0) {
sst.LargestPointKey.CopyFrom(span.LargestKey())
foundPointKey = true
}
}
if !foundPointKey {
// There are no point keys in the span we're interested in.
sst.SmallestPointKey = InternalKey{}
sst.LargestPointKey = InternalKey{}
}
sst.LargestRangeKey.UserKey = sst.LargestRangeKey.UserKey[:0]
sst.LargestRangeKey.Trailer = 0
if rangeKeyIter != nil {
span := rangeKeyIter.SeekLT(upper)
if span != nil {
sst.LargestRangeKey.CopyFrom(span.LargestKey())
} else {
// There are no range keys in the span we're interested in.
sst.SmallestRangeKey = InternalKey{}
sst.LargestRangeKey = InternalKey{}
}
}
}
@@ -584,19 +583,23 @@ func (d *DB) truncateSharedFile(
if len(sst.Smallest.UserKey) == 0 {
return nil, true, nil
}
sst.Size, err = d.tableCache.estimateSize(file, sst.Smallest.UserKey, sst.Largest.UserKey)
if err != nil {
return nil, false, err
}
// On occasion, estimateSize gives us a low estimate, i.e. a 0 file size. This
// can cause panics in places where we divide by file sizes. Correct for it
// here.
if sst.Size == 0 {
sst.Size = 1
}
return sst, false, nil
}
func scanInternalImpl(
ctx context.Context,
lower, upper []byte,
iter *scanInternalIterator,
visitPointKey func(key *InternalKey, value LazyValue) error,
visitRangeDel func(start, end []byte, seqNum uint64) error,
visitRangeKey func(start, end []byte, keys []keyspan.Key) error,
visitSharedFile func(sst *SharedSSTMeta) error,
ctx context.Context, lower, upper []byte, iter *scanInternalIterator, opts *scanInternalOptions,
) error {
if visitSharedFile != nil && (lower == nil || upper == nil) {
if opts.visitSharedFile != nil && (lower == nil || upper == nil) {
panic("lower and upper bounds must be specified in skip-shared iteration mode")
}
// Before starting iteration, check if any files in levels sharedLevelsStart
@@ -605,34 +608,41 @@ func scanInternalImpl(
// of keys. For files that are shared, call visitSharedFile with a truncated
// version of that file.
cmp := iter.comparer.Compare
db := iter.readState.db
provider := db.objProvider
if visitSharedFile != nil {
provider := iter.db.ObjProvider()
seqNum := iter.seqNum
current := iter.version
if current == nil {
current = iter.readState.current
}
if opts.visitSharedFile != nil {
if provider == nil {
panic("expected non-nil Provider in skip-shared iteration mode")
}
for level := sharedLevelsStart; level < numLevels; level++ {
files := iter.readState.current.Levels[level].Iter()
files := current.Levels[level].Iter()
for f := files.SeekGE(cmp, lower); f != nil && cmp(f.Smallest.UserKey, upper) < 0; f = files.Next() {
var objMeta objstorage.ObjectMetadata
var err error
objMeta, err = provider.Lookup(fileTypeTable, f.FileNum.DiskFileNum())
objMeta, err = provider.Lookup(fileTypeTable, f.FileBacking.DiskFileNum)
if err != nil {
return err
}
if !objMeta.IsShared() {
return errors.Wrapf(ErrInvalidSkipSharedIteration, "when processing file %s", objMeta.DiskFileNum)
return errors.Wrapf(ErrInvalidSkipSharedIteration, "file %s is not shared", objMeta.DiskFileNum)
}
if !base.Visible(f.LargestSeqNum, seqNum, base.InternalKeySeqNumMax) {
return errors.Wrapf(ErrInvalidSkipSharedIteration, "file %s contains keys newer than snapshot", objMeta.DiskFileNum)
}
var sst *SharedSSTMeta
var skip bool
sst, skip, err = iter.readState.db.truncateSharedFile(ctx, lower, upper, level, f, objMeta)
sst, skip, err = iter.db.truncateSharedFile(ctx, lower, upper, level, f, objMeta)
if err != nil {
return err
}
if skip {
continue
}
if err = visitSharedFile(sst); err != nil {
if err = opts.visitSharedFile(sst); err != nil {
return err
}
}
@@ -642,21 +652,50 @@ func scanInternalImpl(
for valid := iter.seekGE(lower); valid && iter.error() == nil; valid = iter.next() {
key := iter.unsafeKey()
if opts.rateLimitFunc != nil {
if err := opts.rateLimitFunc(key, iter.lazyValue()); err != nil {
return err
}
}
switch key.Kind() {
case InternalKeyKindRangeKeyDelete, InternalKeyKindRangeKeyUnset, InternalKeyKindRangeKeySet:
span := iter.unsafeSpan()
if err := visitRangeKey(span.Start, span.End, span.Keys); err != nil {
return err
if opts.visitRangeKey != nil {
span := iter.unsafeSpan()
// NB: The caller isn't interested in the sequence numbers of these
// range keys. Rather, the caller wants them to be in trailer order
// _after_ zeroing of sequence numbers. Copy span.Keys, sort it, and then
// call visitRangeKey.
keysCopy := make([]keyspan.Key, len(span.Keys))
for i := range span.Keys {
keysCopy[i] = span.Keys[i]
keysCopy[i].Trailer = base.MakeTrailer(0, span.Keys[i].Kind())
}
keyspan.SortKeysByTrailer(&keysCopy)
if err := opts.visitRangeKey(span.Start, span.End, keysCopy); err != nil {
return err
}
}
case InternalKeyKindRangeDelete:
rangeDel := iter.unsafeRangeDel()
if err := visitRangeDel(rangeDel.Start, rangeDel.End, rangeDel.LargestSeqNum()); err != nil {
return err
if opts.visitRangeDel != nil {
rangeDel := iter.unsafeRangeDel()
if err := opts.visitRangeDel(rangeDel.Start, rangeDel.End, rangeDel.LargestSeqNum()); err != nil {
return err
}
}
default:
val := iter.lazyValue()
if err := visitPointKey(key, val); err != nil {
return err
if opts.visitPointKey != nil {
var info IteratorLevel
if len(iter.mergingIter.heap.items) > 0 {
mergingIterIdx := iter.mergingIter.heap.items[0].index
info = iter.iterLevels[mergingIterIdx]
} else {
info = IteratorLevel{Kind: IteratorLevelUnknown}
}
val := iter.lazyValue()
if err := opts.visitPointKey(key, val, info); err != nil {
return err
}
}
}
}
@@ -676,9 +715,13 @@ func (i *scanInternalIterator) constructPointIter(memtables flushableList, buf *
numMergingLevels := len(memtables)
numLevelIters := 0
current := i.readState.current
current := i.version
if current == nil {
current = i.readState.current
}
numMergingLevels += len(current.L0SublevelFiles)
numLevelIters += len(current.L0SublevelFiles)
for level := 1; level < len(current.Levels); level++ {
if current.Levels[level].Empty() {
continue
@@ -701,29 +744,37 @@ func (i *scanInternalIterator) constructPointIter(memtables flushableList, buf *
rangeDelIters := make([]keyspan.FragmentIterator, 0, numMergingLevels)
rangeDelLevels := make([]keyspan.LevelIter, 0, numLevelIters)
i.iterLevels = make([]IteratorLevel, numMergingLevels)
mlevelsIndex := 0
// Next are the memtables.
for j := len(memtables) - 1; j >= 0; j-- {
mem := memtables[j]
mlevels = append(mlevels, mergingIterLevel{
iter: mem.newIter(&i.opts.IterOptions),
})
i.iterLevels[mlevelsIndex] = IteratorLevel{
Kind: IteratorLevelFlushable,
FlushableIndex: j,
}
mlevelsIndex++
if rdi := mem.newRangeDelIter(&i.opts.IterOptions); rdi != nil {
rangeDelIters = append(rangeDelIters, rdi)
}
}
// Next are the file levels: L0 sub-levels followed by lower levels.
mlevelsIndex := len(mlevels)
levelsIndex := len(levels)
mlevels = mlevels[:numMergingLevels]
levels = levels[:numLevelIters]
rangeDelLevels = rangeDelLevels[:numLevelIters]
i.opts.IterOptions.snapshotForHideObsoletePoints = i.seqNum
addLevelIterForFiles := func(files manifest.LevelIterator, level manifest.Level) {
li := &levels[levelsIndex]
rli := &rangeDelLevels[levelsIndex]
li.init(
context.Background(), i.opts.IterOptions, i.comparer.Compare, i.comparer.Split, i.newIters, files, level,
context.Background(), i.opts.IterOptions, i.comparer, i.newIters, files, level,
internalIterOpts{})
li.initBoundaryContext(&mlevels[mlevelsIndex].levelIterBoundaryContext)
mlevels[mlevelsIndex].iter = li
@@ -736,12 +787,14 @@ func (i *scanInternalIterator) constructPointIter(memtables flushableList, buf *
mlevelsIndex++
}
// Add level iterators for the L0 sublevels, iterating from newest to
// oldest.
for i := len(current.L0SublevelFiles) - 1; i >= 0; i-- {
addLevelIterForFiles(current.L0SublevelFiles[i].Iter(), manifest.L0Sublevel(i))
for j := len(current.L0SublevelFiles) - 1; j >= 0; j-- {
i.iterLevels[mlevelsIndex] = IteratorLevel{
Kind: IteratorLevelLSM,
Level: 0,
Sublevel: j,
}
addLevelIterForFiles(current.L0SublevelFiles[j].Iter(), manifest.L0Sublevel(j))
}
// Add level iterators for the non-empty non-L0 levels.
for level := 1; level < numLevels; level++ {
if current.Levels[level].Empty() {
@@ -750,18 +803,35 @@ func (i *scanInternalIterator) constructPointIter(memtables flushableList, buf *
if i.opts.skipSharedLevels && level >= sharedLevelsStart {
continue
}
i.iterLevels[mlevelsIndex] = IteratorLevel{Kind: IteratorLevelLSM, Level: level}
addLevelIterForFiles(current.Levels[level].Iter(), manifest.Level(level))
}
buf.merging.init(&i.opts.IterOptions, &InternalIteratorStats{}, i.comparer.Compare, i.comparer.Split, mlevels...)
buf.merging.snapshot = i.seqNum
rangeDelMiter.Init(i.comparer.Compare, keyspan.VisibleTransform(i.seqNum), new(keyspan.MergingBuffers), rangeDelIters...)
i.pointKeyIter = pointCollapsingIterator{
comparer: i.comparer,
merge: i.merge,
seqNum: i.seqNum,
if i.opts.includeObsoleteKeys {
iiter := &keyspan.InterleavingIter{}
iiter.Init(i.comparer, &buf.merging, &rangeDelMiter,
keyspan.InterleavingIterOpts{
LowerBound: i.opts.LowerBound,
UpperBound: i.opts.UpperBound,
})
i.pointKeyIter = iiter
} else {
pcIter := &pointCollapsingIterator{
comparer: i.comparer,
merge: i.merge,
seqNum: i.seqNum,
}
pcIter.iter.Init(i.comparer, &buf.merging, &rangeDelMiter, keyspan.InterleavingIterOpts{
LowerBound: i.opts.LowerBound,
UpperBound: i.opts.UpperBound,
})
i.pointKeyIter = pcIter
}
i.pointKeyIter.iter.Init(i.comparer, &buf.merging, &rangeDelMiter, nil /* mask */, i.opts.LowerBound, i.opts.UpperBound)
i.iter = &i.pointKeyIter
i.iter = i.pointKeyIter
}
// constructRangeKeyIter constructs the range-key iterator stack, populating
@@ -773,23 +843,28 @@ func (i *scanInternalIterator) constructRangeKeyIter() {
// RangeKeyUnsets and RangeKeyDels.
i.rangeKey.rangeKeyIter = i.rangeKey.iterConfig.Init(
i.comparer, i.seqNum, i.opts.LowerBound, i.opts.UpperBound,
nil /* hasPrefix */, nil /* prefix */, false, /* onlySets */
nil /* hasPrefix */, nil /* prefix */, true, /* internalKeys */
&i.rangeKey.rangeKeyBuffers.internal)
// Next are the flushables: memtables and large batches.
for j := len(i.readState.memtables) - 1; j >= 0; j-- {
mem := i.readState.memtables[j]
// We only need to read from memtables which contain sequence numbers older
// than seqNum.
if logSeqNum := mem.logSeqNum; logSeqNum >= i.seqNum {
continue
}
if rki := mem.newRangeKeyIter(&i.opts.IterOptions); rki != nil {
i.rangeKey.iterConfig.AddLevel(rki)
if i.readState != nil {
for j := len(i.readState.memtables) - 1; j >= 0; j-- {
mem := i.readState.memtables[j]
// We only need to read from memtables which contain sequence numbers older
// than seqNum.
if logSeqNum := mem.logSeqNum; logSeqNum >= i.seqNum {
continue
}
if rki := mem.newRangeKeyIter(&i.opts.IterOptions); rki != nil {
i.rangeKey.iterConfig.AddLevel(rki)
}
}
}
current := i.readState.current
current := i.version
if current == nil {
current = i.readState.current
}
// Next are the file levels: L0 sub-levels followed by lower levels.
//
// Add file-specific iterators for L0 files containing range keys. This is less
@@ -804,8 +879,7 @@ func (i *scanInternalIterator) constructRangeKeyIter() {
// around Key Trailer order.
iter := current.RangeKeyLevels[0].Iter()
for f := iter.Last(); f != nil; f = iter.Prev() {
spanIterOpts := &keyspan.SpanIterOptions{RangeKeyFilters: i.opts.RangeKeyFilters}
spanIter, err := i.newIterRangeKey(f, spanIterOpts)
spanIter, err := i.newIterRangeKey(f, i.opts.SpanIterOptions())
if err != nil {
i.rangeKey.iterConfig.AddLevel(&errorKeyspanIter{err: err})
continue
@@ -822,7 +896,7 @@ func (i *scanInternalIterator) constructRangeKeyIter() {
continue
}
li := i.rangeKey.iterConfig.NewLevelIter()
spanIterOpts := keyspan.SpanIterOptions{RangeKeyFilters: i.opts.RangeKeyFilters}
spanIterOpts := i.opts.SpanIterOptions()
li.Init(spanIterOpts, i.comparer.Compare, i.newIterRangeKey, current.RangeKeyLevels[level].Iter(),
manifest.Level(level), manifest.KeyTypeRange)
i.rangeKey.iterConfig.AddLevel(li)
@@ -852,7 +926,10 @@ func (i *scanInternalIterator) lazyValue() LazyValue {
// unsafeRangeDel returns a range key span. Behaviour undefined if UnsafeKey returns
// a non-rangedel kind.
func (i *scanInternalIterator) unsafeRangeDel() *keyspan.Span {
return i.pointKeyIter.iter.Span()
type spanInternalIterator interface {
Span() *keyspan.Span
}
return i.pointKeyIter.(spanInternalIterator).Span()
}
// unsafeSpan returns a range key span. Behaviour undefined if UnsafeKey returns
@@ -878,7 +955,12 @@ func (i *scanInternalIterator) close() error {
if err := i.iter.Close(); err != nil {
return err
}
i.readState.unref()
if i.readState != nil {
i.readState.unref()
}
if i.version != nil {
i.version.Unref()
}
if i.rangeKey != nil {
i.rangeKey.PrepareForReuse()
*i.rangeKey = iteratorRangeKeyState{