Line data Source code
1 : // Copyright 2023 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 :
11 : "github.com/cockroachdb/errors"
12 : "github.com/cockroachdb/pebble/internal/base"
13 : "github.com/cockroachdb/pebble/internal/invariants"
14 : "github.com/cockroachdb/pebble/internal/keyspan"
15 : "github.com/cockroachdb/pebble/internal/keyspan/keyspanimpl"
16 : "github.com/cockroachdb/pebble/internal/manifest"
17 : "github.com/cockroachdb/pebble/objstorage"
18 : "github.com/cockroachdb/pebble/objstorage/remote"
19 : "github.com/cockroachdb/pebble/sstable"
20 : )
21 :
22 : const (
23 : // In skip-shared iteration mode, keys in levels sharedLevelsStart and greater
24 : // (i.e. lower in the LSM) are skipped.
25 : sharedLevelsStart = remote.SharedLevelsStart
26 : )
27 :
28 : // ErrInvalidSkipSharedIteration is returned by ScanInternal if it was called
29 : // with a shared file visitor function, and a file in a shareable level (i.e.
30 : // level >= sharedLevelsStart) was found to not be in shared storage according
31 : // to objstorage.Provider, or not shareable for another reason such as for
32 : // containing keys newer than the snapshot sequence number.
33 : var ErrInvalidSkipSharedIteration = errors.New("pebble: cannot use skip-shared iteration due to non-shareable files in lower levels")
34 :
35 : // SharedSSTMeta represents an sstable on shared storage that can be ingested
36 : // by another pebble instance. This struct must contain all fields that are
37 : // required for a Pebble instance to ingest a foreign sstable on shared storage,
38 : // including constructing any relevant objstorage.Provider / remoteobjcat.Catalog
39 : // data structures, as well as creating virtual FileMetadatas.
40 : //
41 : // Note that the Pebble instance creating and returning a SharedSSTMeta might
42 : // not be the one that created the underlying sstable on shared storage to begin
43 : // with; it's possible for a Pebble instance to reshare an sstable that was
44 : // shared to it.
45 : type SharedSSTMeta struct {
46 : // Backing is the shared object underlying this SST. Can be attached to an
47 : // objstorage.Provider.
48 : Backing objstorage.RemoteObjectBackingHandle
49 :
50 : // Smallest and Largest internal keys for the overall bounds. The kind and
51 : // SeqNum of these will reflect what is physically present on the source Pebble
52 : // instance's view of the sstable; it's up to the ingesting instance to set the
53 : // sequence number in the trailer to match the read-time sequence numbers
54 : // reserved for the level this SST is being ingested into. The Kind is expected
55 : // to remain unchanged by the ingesting instance.
56 : //
57 : // Note that these bounds could be narrower than the bounds of the underlying
58 : // sstable; ScanInternal is expected to truncate sstable bounds to the user key
59 : // bounds passed into that method.
60 : Smallest, Largest InternalKey
61 :
62 : // SmallestRangeKey and LargestRangeKey are internal keys that denote the
63 : // range key bounds of this sstable. Must lie within [Smallest, Largest].
64 : SmallestRangeKey, LargestRangeKey InternalKey
65 :
66 : // SmallestPointKey and LargestPointKey are internal keys that denote the
67 : // point key bounds of this sstable. Must lie within [Smallest, Largest].
68 : SmallestPointKey, LargestPointKey InternalKey
69 :
70 : // Level denotes the level at which this file was present at read time.
71 : // For files visited by ScanInternal, this value will only be 5 or 6.
72 : Level uint8
73 :
74 : // Size contains an estimate of the size of this sstable.
75 : Size uint64
76 :
77 : // fileNum at time of creation in the creator instance. Only used for
78 : // debugging/tests.
79 : fileNum base.FileNum
80 : }
81 :
82 2 : func (s *SharedSSTMeta) cloneFromFileMeta(f *fileMetadata) {
83 2 : *s = SharedSSTMeta{
84 2 : Smallest: f.Smallest.Clone(),
85 2 : Largest: f.Largest.Clone(),
86 2 : SmallestRangeKey: f.SmallestRangeKey.Clone(),
87 2 : LargestRangeKey: f.LargestRangeKey.Clone(),
88 2 : SmallestPointKey: f.SmallestPointKey.Clone(),
89 2 : LargestPointKey: f.LargestPointKey.Clone(),
90 2 : Size: f.Size,
91 2 : fileNum: f.FileNum,
92 2 : }
93 2 : }
94 :
95 : type sharedByLevel []SharedSSTMeta
96 :
97 2 : func (s sharedByLevel) Len() int { return len(s) }
98 0 : func (s sharedByLevel) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
99 2 : func (s sharedByLevel) Less(i, j int) bool { return s[i].Level < s[j].Level }
100 :
101 : type pcIterPos int
102 :
103 : const (
104 : pcIterPosCur pcIterPos = iota
105 : pcIterPosNext
106 : )
107 :
108 : // pointCollapsingIterator is an internalIterator that collapses point keys and
109 : // returns at most one point internal key for each user key. Merges and
110 : // SingleDels are not supported and result in a panic if encountered. Point keys
111 : // deleted by rangedels are considered shadowed and not exposed.
112 : //
113 : // Only used in ScanInternal to return at most one internal key per user key.
114 : type pointCollapsingIterator struct {
115 : iter keyspan.InterleavingIter
116 : pos pcIterPos
117 : comparer *base.Comparer
118 : merge base.Merge
119 : err error
120 : seqNum uint64
121 : // The current position of `iter`. Always owned by the underlying iter.
122 : iterKey *InternalKey
123 : // The last saved key. findNextEntry and similar methods are expected to save
124 : // the current value of iterKey to savedKey if they're iterating away from the
125 : // current key but still need to retain it. See comments in findNextEntry on
126 : // how this field is used.
127 : //
128 : // At the end of a positioning call:
129 : // - if pos == pcIterPosNext, iterKey is pointing to the next user key owned
130 : // by `iter` while savedKey is holding a copy to our current key.
131 : // - If pos == pcIterPosCur, iterKey is pointing to an `iter`-owned current
132 : // key, and savedKey is either undefined or pointing to a version of the
133 : // current key owned by this iterator (i.e. backed by savedKeyBuf).
134 : savedKey InternalKey
135 : savedKeyBuf []byte
136 : // Value at the current iterator position, at iterKey.
137 : iterValue base.LazyValue
138 : // If fixedSeqNum is non-zero, all emitted points are verified to have this
139 : // fixed sequence number.
140 : fixedSeqNum uint64
141 : }
142 :
143 2 : func (p *pointCollapsingIterator) Span() *keyspan.Span {
144 2 : return p.iter.Span()
145 2 : }
146 :
147 : // SeekPrefixGE implements the InternalIterator interface.
148 : func (p *pointCollapsingIterator) SeekPrefixGE(
149 : prefix, key []byte, flags base.SeekGEFlags,
150 0 : ) (*base.InternalKey, base.LazyValue) {
151 0 : p.resetKey()
152 0 : p.iterKey, p.iterValue = p.iter.SeekPrefixGE(prefix, key, flags)
153 0 : p.pos = pcIterPosCur
154 0 : if p.iterKey == nil {
155 0 : return nil, base.LazyValue{}
156 0 : }
157 0 : return p.findNextEntry()
158 : }
159 :
160 : // SeekGE implements the InternalIterator interface.
161 : func (p *pointCollapsingIterator) SeekGE(
162 : key []byte, flags base.SeekGEFlags,
163 2 : ) (*base.InternalKey, base.LazyValue) {
164 2 : p.resetKey()
165 2 : p.iterKey, p.iterValue = p.iter.SeekGE(key, flags)
166 2 : p.pos = pcIterPosCur
167 2 : if p.iterKey == nil {
168 2 : return nil, base.LazyValue{}
169 2 : }
170 2 : return p.findNextEntry()
171 : }
172 :
173 : // SeekLT implements the InternalIterator interface.
174 : func (p *pointCollapsingIterator) SeekLT(
175 : key []byte, flags base.SeekLTFlags,
176 0 : ) (*base.InternalKey, base.LazyValue) {
177 0 : panic("unimplemented")
178 : }
179 :
180 2 : func (p *pointCollapsingIterator) resetKey() {
181 2 : p.savedKey.UserKey = p.savedKeyBuf[:0]
182 2 : p.savedKey.Trailer = 0
183 2 : p.iterKey = nil
184 2 : p.pos = pcIterPosCur
185 2 : }
186 :
187 2 : func (p *pointCollapsingIterator) verifySeqNum(key *base.InternalKey) *base.InternalKey {
188 2 : if !invariants.Enabled {
189 0 : return key
190 0 : }
191 2 : if p.fixedSeqNum == 0 || key == nil || key.Kind() == InternalKeyKindRangeDelete {
192 2 : return key
193 2 : }
194 0 : if key.SeqNum() != p.fixedSeqNum {
195 0 : panic(fmt.Sprintf("expected foreign point key to have seqnum %d, got %d", p.fixedSeqNum, key.SeqNum()))
196 : }
197 0 : return key
198 : }
199 :
200 : // findNextEntry is called to return the next key. p.iter must be positioned at the
201 : // start of the first user key we are interested in.
202 2 : func (p *pointCollapsingIterator) findNextEntry() (*base.InternalKey, base.LazyValue) {
203 2 : p.saveKey()
204 2 : // Saves a comparison in the fast path
205 2 : firstIteration := true
206 2 : for p.iterKey != nil {
207 2 : // NB: p.savedKey is either the current key (iff p.iterKey == firstKey),
208 2 : // or the previous key.
209 2 : if !firstIteration && !p.comparer.Equal(p.iterKey.UserKey, p.savedKey.UserKey) {
210 2 : p.saveKey()
211 2 : continue
212 : }
213 2 : firstIteration = false
214 2 : if s := p.iter.Span(); s != nil && s.CoversAt(p.seqNum, p.iterKey.SeqNum()) {
215 2 : // All future keys for this user key must be deleted.
216 2 : if p.savedKey.Kind() == InternalKeyKindSingleDelete {
217 0 : panic("cannot process singledel key in point collapsing iterator")
218 : }
219 : // Fast forward to the next user key.
220 2 : p.saveKey()
221 2 : p.iterKey, p.iterValue = p.iter.Next()
222 2 : for p.iterKey != nil && p.savedKey.SeqNum() >= p.iterKey.SeqNum() && p.comparer.Equal(p.iterKey.UserKey, p.savedKey.UserKey) {
223 2 : p.iterKey, p.iterValue = p.iter.Next()
224 2 : }
225 2 : continue
226 : }
227 2 : switch p.savedKey.Kind() {
228 2 : case InternalKeyKindSet, InternalKeyKindDelete, InternalKeyKindSetWithDelete, InternalKeyKindDeleteSized:
229 2 : // Note that we return SETs directly, even if they would otherwise get
230 2 : // compacted into a Del to turn into a SetWithDelete. This is a fast
231 2 : // path optimization that can break SINGLEDEL determinism. To lead to
232 2 : // consistent SINGLEDEL behaviour, this iterator should *not* be used for
233 2 : // a keyspace where SINGLEDELs could be in use. If this iterator observes
234 2 : // a SINGLEDEL as the first internal key for a user key, it will panic.
235 2 : //
236 2 : // As p.value is a lazy value owned by the child iterator, we can thread
237 2 : // it through without loading it into p.valueBuf.
238 2 : //
239 2 : // TODO(bilal): We can even avoid saving the key in this fast path if
240 2 : // we are in a block where setHasSamePrefix = false in a v3 sstable,
241 2 : // guaranteeing that there's only one internal key for each user key.
242 2 : // Thread this logic through the sstable iterators and/or consider
243 2 : // collapsing (ha) this logic into the sstable iterators that are aware
244 2 : // of blocks and can determine user key changes without doing key saves
245 2 : // or comparisons.
246 2 : p.pos = pcIterPosCur
247 2 : return p.verifySeqNum(p.iterKey), p.iterValue
248 0 : case InternalKeyKindSingleDelete:
249 0 : // Panic, as this iterator is not expected to observe single deletes.
250 0 : panic("cannot process singledel key in point collapsing iterator")
251 0 : case InternalKeyKindMerge:
252 0 : // Panic, as this iterator is not expected to observe merges.
253 0 : panic("cannot process merge key in point collapsing iterator")
254 2 : case InternalKeyKindRangeDelete:
255 2 : // These are interleaved by the interleaving iterator ahead of all points.
256 2 : // We should pass them as-is, but also account for any points ahead of
257 2 : // them.
258 2 : p.pos = pcIterPosCur
259 2 : return p.verifySeqNum(p.iterKey), p.iterValue
260 0 : default:
261 0 : panic(fmt.Sprintf("unexpected kind: %d", p.iterKey.Kind()))
262 : }
263 : }
264 1 : p.resetKey()
265 1 : return nil, base.LazyValue{}
266 : }
267 :
268 : // First implements the InternalIterator interface.
269 1 : func (p *pointCollapsingIterator) First() (*base.InternalKey, base.LazyValue) {
270 1 : p.resetKey()
271 1 : p.iterKey, p.iterValue = p.iter.First()
272 1 : p.pos = pcIterPosCur
273 1 : if p.iterKey == nil {
274 0 : return nil, base.LazyValue{}
275 0 : }
276 1 : return p.findNextEntry()
277 : }
278 :
279 : // Last implements the InternalIterator interface.
280 0 : func (p *pointCollapsingIterator) Last() (*base.InternalKey, base.LazyValue) {
281 0 : panic("unimplemented")
282 : }
283 :
284 2 : func (p *pointCollapsingIterator) saveKey() {
285 2 : if p.iterKey == nil {
286 1 : p.savedKey = InternalKey{UserKey: p.savedKeyBuf[:0]}
287 1 : return
288 1 : }
289 2 : p.savedKeyBuf = append(p.savedKeyBuf[:0], p.iterKey.UserKey...)
290 2 : p.savedKey = InternalKey{UserKey: p.savedKeyBuf, Trailer: p.iterKey.Trailer}
291 : }
292 :
293 : // Next implements the InternalIterator interface.
294 2 : func (p *pointCollapsingIterator) Next() (*base.InternalKey, base.LazyValue) {
295 2 : switch p.pos {
296 2 : case pcIterPosCur:
297 2 : p.saveKey()
298 2 : if p.iterKey != nil && p.iterKey.Kind() == InternalKeyKindRangeDelete {
299 2 : // Step over the interleaved range delete and process the very next
300 2 : // internal key, even if it's at the same user key. This is because a
301 2 : // point for that user key has not been returned yet.
302 2 : p.iterKey, p.iterValue = p.iter.Next()
303 2 : break
304 : }
305 : // Fast forward to the next user key.
306 2 : key, val := p.iter.Next()
307 2 : // p.iterKey.SeqNum() >= key.SeqNum() is an optimization that allows us to
308 2 : // use p.iterKey.SeqNum() < key.SeqNum() as a sign that the user key has
309 2 : // changed, without needing to do the full key comparison.
310 2 : for key != nil && p.savedKey.SeqNum() >= key.SeqNum() &&
311 2 : p.comparer.Equal(p.savedKey.UserKey, key.UserKey) {
312 2 : key, val = p.iter.Next()
313 2 : }
314 2 : if key == nil {
315 2 : // There are no keys to return.
316 2 : p.resetKey()
317 2 : return nil, base.LazyValue{}
318 2 : }
319 2 : p.iterKey, p.iterValue = key, val
320 0 : case pcIterPosNext:
321 0 : p.pos = pcIterPosCur
322 : }
323 2 : if p.iterKey == nil {
324 2 : p.resetKey()
325 2 : return nil, base.LazyValue{}
326 2 : }
327 2 : return p.findNextEntry()
328 : }
329 :
330 : // NextPrefix implements the InternalIterator interface.
331 0 : func (p *pointCollapsingIterator) NextPrefix(succKey []byte) (*base.InternalKey, base.LazyValue) {
332 0 : panic("unimplemented")
333 : }
334 :
335 : // Prev implements the InternalIterator interface.
336 0 : func (p *pointCollapsingIterator) Prev() (*base.InternalKey, base.LazyValue) {
337 0 : panic("unimplemented")
338 : }
339 :
340 : // Error implements the InternalIterator interface.
341 2 : func (p *pointCollapsingIterator) Error() error {
342 2 : if p.err != nil {
343 0 : return p.err
344 0 : }
345 2 : return p.iter.Error()
346 : }
347 :
348 : // Close implements the InternalIterator interface.
349 2 : func (p *pointCollapsingIterator) Close() error {
350 2 : return p.iter.Close()
351 2 : }
352 :
353 : // SetBounds implements the InternalIterator interface.
354 0 : func (p *pointCollapsingIterator) SetBounds(lower, upper []byte) {
355 0 : p.resetKey()
356 0 : p.iter.SetBounds(lower, upper)
357 0 : }
358 :
359 0 : func (p *pointCollapsingIterator) SetContext(ctx context.Context) {
360 0 : p.iter.SetContext(ctx)
361 0 : }
362 :
363 : // String implements the InternalIterator interface.
364 0 : func (p *pointCollapsingIterator) String() string {
365 0 : return p.iter.String()
366 0 : }
367 :
368 : var _ internalIterator = &pointCollapsingIterator{}
369 :
370 : // IteratorLevelKind is used to denote whether the current ScanInternal iterator
371 : // is unknown, belongs to a flushable, or belongs to an LSM level type.
372 : type IteratorLevelKind int8
373 :
374 : const (
375 : // IteratorLevelUnknown indicates an unknown LSM level.
376 : IteratorLevelUnknown IteratorLevelKind = iota
377 : // IteratorLevelLSM indicates an LSM level.
378 : IteratorLevelLSM
379 : // IteratorLevelFlushable indicates a flushable (i.e. memtable).
380 : IteratorLevelFlushable
381 : )
382 :
383 : // IteratorLevel is used with scanInternalIterator to surface additional iterator-specific info where possible.
384 : // Note: this is struct is only provided for point keys.
385 : type IteratorLevel struct {
386 : Kind IteratorLevelKind
387 : // FlushableIndex indicates the position within the flushable queue of this level.
388 : // Only valid if kind == IteratorLevelFlushable.
389 : FlushableIndex int
390 : // The level within the LSM. Only valid if Kind == IteratorLevelLSM.
391 : Level int
392 : // Sublevel is only valid if Kind == IteratorLevelLSM and Level == 0.
393 : Sublevel int
394 : }
395 :
396 : // scanInternalIterator is an iterator that returns all internal keys, including
397 : // tombstones. For instance, an InternalKeyKindDelete would be returned as an
398 : // InternalKeyKindDelete instead of the iterator skipping over to the next key.
399 : // Internal keys within a user key are collapsed, eg. if there are two SETs, the
400 : // one with the higher sequence is returned. Useful if an external user of Pebble
401 : // needs to observe and rebuild Pebble's history of internal keys, such as in
402 : // node-to-node replication. For use with {db,snapshot}.ScanInternal().
403 : //
404 : // scanInternalIterator is expected to ignore point keys deleted by range
405 : // deletions, and range keys shadowed by a range key unset or delete, however it
406 : // *must* return the range delete as well as the range key unset/delete that did
407 : // the shadowing.
408 : type scanInternalIterator struct {
409 : ctx context.Context
410 : db *DB
411 : opts scanInternalOptions
412 : comparer *base.Comparer
413 : merge Merge
414 : iter internalIterator
415 : readState *readState
416 : version *version
417 : rangeKey *iteratorRangeKeyState
418 : pointKeyIter internalIterator
419 : iterKey *InternalKey
420 : iterValue LazyValue
421 : alloc *iterAlloc
422 : newIters tableNewIters
423 : newIterRangeKey keyspanimpl.TableNewSpanIter
424 : seqNum uint64
425 : iterLevels []IteratorLevel
426 : mergingIter *mergingIter
427 :
428 : // boundsBuf holds two buffers used to store the lower and upper bounds.
429 : // Whenever the InternalIterator's bounds change, the new bounds are copied
430 : // into boundsBuf[boundsBufIdx]. The two bounds share a slice to reduce
431 : // allocations. opts.LowerBound and opts.UpperBound point into this slice.
432 : boundsBuf [2][]byte
433 : boundsBufIdx int
434 : }
435 :
436 : // truncateSharedFile truncates a shared file's [Smallest, Largest] fields to
437 : // [lower, upper), potentially opening iterators on the file to find keys within
438 : // the requested bounds. A SharedSSTMeta is produced that is suitable for
439 : // external consumption by other Pebble instances. If shouldSkip is true, this
440 : // file does not contain any keys in [lower, upper) and can be skipped.
441 : //
442 : // TODO(bilal): If opening iterators and doing reads in this method is too
443 : // inefficient, consider producing non-tight file bounds instead.
444 : func (d *DB) truncateSharedFile(
445 : ctx context.Context,
446 : lower, upper []byte,
447 : level int,
448 : file *fileMetadata,
449 : objMeta objstorage.ObjectMetadata,
450 2 : ) (sst *SharedSSTMeta, shouldSkip bool, err error) {
451 2 : cmp := d.cmp
452 2 : sst = &SharedSSTMeta{}
453 2 : sst.cloneFromFileMeta(file)
454 2 : sst.Level = uint8(level)
455 2 : sst.Backing, err = d.objProvider.RemoteObjectBacking(&objMeta)
456 2 : if err != nil {
457 0 : return nil, false, err
458 0 : }
459 2 : needsLowerTruncate := cmp(lower, file.Smallest.UserKey) > 0
460 2 : needsUpperTruncate := cmp(upper, file.Largest.UserKey) < 0 || (cmp(upper, file.Largest.UserKey) == 0 && !file.Largest.IsExclusiveSentinel())
461 2 : // Fast path: file is entirely within [lower, upper).
462 2 : if !needsLowerTruncate && !needsUpperTruncate {
463 2 : return sst, false, nil
464 2 : }
465 :
466 : // We will need to truncate file bounds in at least one direction. Open all
467 : // relevant iterators.
468 2 : iter, rangeDelIter, err := d.newIters.TODO(ctx, file, &IterOptions{
469 2 : LowerBound: lower,
470 2 : UpperBound: upper,
471 2 : level: manifest.Level(level),
472 2 : }, internalIterOpts{})
473 2 : if err != nil {
474 0 : return nil, false, err
475 0 : }
476 2 : defer iter.Close()
477 2 : if rangeDelIter != nil {
478 2 : rangeDelIter = keyspan.Truncate(
479 2 : cmp, rangeDelIter, lower, upper, nil, nil,
480 2 : false, /* panicOnUpperTruncate */
481 2 : )
482 2 : defer rangeDelIter.Close()
483 2 : }
484 2 : rangeKeyIter, err := d.tableNewRangeKeyIter(file, keyspan.SpanIterOptions{})
485 2 : if err != nil {
486 0 : return nil, false, err
487 0 : }
488 2 : if rangeKeyIter != nil {
489 2 : rangeKeyIter = keyspan.Truncate(
490 2 : cmp, rangeKeyIter, lower, upper, nil, nil,
491 2 : false, /* panicOnUpperTruncate */
492 2 : )
493 2 : defer rangeKeyIter.Close()
494 2 : }
495 : // Check if we need to truncate on the left side. This means finding a new
496 : // LargestPointKey and LargestRangeKey that is >= lower.
497 2 : if needsLowerTruncate {
498 2 : sst.SmallestPointKey.UserKey = sst.SmallestPointKey.UserKey[:0]
499 2 : sst.SmallestPointKey.Trailer = 0
500 2 : key, _ := iter.SeekGE(lower, base.SeekGEFlagsNone)
501 2 : foundPointKey := key != nil
502 2 : if key != nil {
503 2 : sst.SmallestPointKey.CopyFrom(*key)
504 2 : }
505 2 : if rangeDelIter != nil {
506 2 : if span, err := rangeDelIter.SeekGE(lower); err != nil {
507 0 : return nil, false, err
508 2 : } else if span != nil && (len(sst.SmallestPointKey.UserKey) == 0 || base.InternalCompare(cmp, span.SmallestKey(), sst.SmallestPointKey) < 0) {
509 2 : sst.SmallestPointKey.CopyFrom(span.SmallestKey())
510 2 : foundPointKey = true
511 2 : }
512 : }
513 2 : if !foundPointKey {
514 2 : // There are no point keys in the span we're interested in.
515 2 : sst.SmallestPointKey = InternalKey{}
516 2 : sst.LargestPointKey = InternalKey{}
517 2 : }
518 2 : sst.SmallestRangeKey.UserKey = sst.SmallestRangeKey.UserKey[:0]
519 2 : sst.SmallestRangeKey.Trailer = 0
520 2 : if rangeKeyIter != nil {
521 2 : span, err := rangeKeyIter.SeekGE(lower)
522 2 : switch {
523 0 : case err != nil:
524 0 : return nil, false, err
525 2 : case span != nil:
526 2 : sst.SmallestRangeKey.CopyFrom(span.SmallestKey())
527 2 : default:
528 2 : // There are no range keys in the span we're interested in.
529 2 : sst.SmallestRangeKey = InternalKey{}
530 2 : sst.LargestRangeKey = InternalKey{}
531 : }
532 : }
533 : }
534 : // Check if we need to truncate on the right side. This means finding a new
535 : // LargestPointKey and LargestRangeKey that is < upper.
536 2 : if needsUpperTruncate {
537 2 : sst.LargestPointKey.UserKey = sst.LargestPointKey.UserKey[:0]
538 2 : sst.LargestPointKey.Trailer = 0
539 2 : key, _ := iter.SeekLT(upper, base.SeekLTFlagsNone)
540 2 : foundPointKey := key != nil
541 2 : if key != nil {
542 2 : sst.LargestPointKey.CopyFrom(*key)
543 2 : }
544 2 : if rangeDelIter != nil {
545 2 : if span, err := rangeDelIter.SeekLT(upper); err != nil {
546 0 : return nil, false, err
547 2 : } else if span != nil && (len(sst.LargestPointKey.UserKey) == 0 || base.InternalCompare(cmp, span.LargestKey(), sst.LargestPointKey) > 0) {
548 2 : sst.LargestPointKey.CopyFrom(span.LargestKey())
549 2 : foundPointKey = true
550 2 : }
551 : }
552 2 : if !foundPointKey {
553 2 : // There are no point keys in the span we're interested in.
554 2 : sst.SmallestPointKey = InternalKey{}
555 2 : sst.LargestPointKey = InternalKey{}
556 2 : }
557 2 : sst.LargestRangeKey.UserKey = sst.LargestRangeKey.UserKey[:0]
558 2 : sst.LargestRangeKey.Trailer = 0
559 2 : if rangeKeyIter != nil {
560 2 : span, err := rangeKeyIter.SeekLT(upper)
561 2 : switch {
562 0 : case err != nil:
563 0 : return nil, false, err
564 2 : case span != nil:
565 2 : sst.LargestRangeKey.CopyFrom(span.LargestKey())
566 2 : default:
567 2 : // There are no range keys in the span we're interested in.
568 2 : sst.SmallestRangeKey = InternalKey{}
569 2 : sst.LargestRangeKey = InternalKey{}
570 : }
571 : }
572 : }
573 : // Set overall bounds based on {Smallest,Largest}{Point,Range}Key.
574 2 : switch {
575 2 : case len(sst.SmallestRangeKey.UserKey) == 0:
576 2 : sst.Smallest = sst.SmallestPointKey
577 2 : case len(sst.SmallestPointKey.UserKey) == 0:
578 2 : sst.Smallest = sst.SmallestRangeKey
579 2 : default:
580 2 : sst.Smallest = sst.SmallestPointKey
581 2 : if base.InternalCompare(cmp, sst.SmallestRangeKey, sst.SmallestPointKey) < 0 {
582 2 : sst.Smallest = sst.SmallestRangeKey
583 2 : }
584 : }
585 2 : switch {
586 2 : case len(sst.LargestRangeKey.UserKey) == 0:
587 2 : sst.Largest = sst.LargestPointKey
588 2 : case len(sst.LargestPointKey.UserKey) == 0:
589 2 : sst.Largest = sst.LargestRangeKey
590 2 : default:
591 2 : sst.Largest = sst.LargestPointKey
592 2 : if base.InternalCompare(cmp, sst.LargestRangeKey, sst.LargestPointKey) > 0 {
593 2 : sst.Largest = sst.LargestRangeKey
594 2 : }
595 : }
596 : // On rare occasion, a file might overlap with [lower, upper) but not actually
597 : // have any keys within those bounds. Skip such files.
598 2 : if len(sst.Smallest.UserKey) == 0 {
599 2 : return nil, true, nil
600 2 : }
601 2 : sst.Size, err = d.tableCache.estimateSize(file, sst.Smallest.UserKey, sst.Largest.UserKey)
602 2 : if err != nil {
603 0 : return nil, false, err
604 0 : }
605 : // On occasion, estimateSize gives us a low estimate, i.e. a 0 file size. This
606 : // can cause panics in places where we divide by file sizes. Correct for it
607 : // here.
608 2 : if sst.Size == 0 {
609 2 : sst.Size = 1
610 2 : }
611 2 : return sst, false, nil
612 : }
613 :
614 : func scanInternalImpl(
615 : ctx context.Context, lower, upper []byte, iter *scanInternalIterator, opts *scanInternalOptions,
616 2 : ) error {
617 2 : if opts.visitSharedFile != nil && (lower == nil || upper == nil) {
618 0 : panic("lower and upper bounds must be specified in skip-shared iteration mode")
619 : }
620 : // Before starting iteration, check if any files in levels sharedLevelsStart
621 : // and below are *not* shared. Error out if that is the case, as skip-shared
622 : // iteration will not produce a consistent point-in-time view of this range
623 : // of keys. For files that are shared, call visitSharedFile with a truncated
624 : // version of that file.
625 2 : cmp := iter.comparer.Compare
626 2 : provider := iter.db.ObjProvider()
627 2 : seqNum := iter.seqNum
628 2 : current := iter.version
629 2 : if current == nil {
630 2 : current = iter.readState.current
631 2 : }
632 2 : if opts.visitSharedFile != nil {
633 2 : if provider == nil {
634 0 : panic("expected non-nil Provider in skip-shared iteration mode")
635 : }
636 2 : for level := sharedLevelsStart; level < numLevels; level++ {
637 2 : files := current.Levels[level].Iter()
638 2 : for f := files.SeekGE(cmp, lower); f != nil && cmp(f.Smallest.UserKey, upper) < 0; f = files.Next() {
639 2 : var objMeta objstorage.ObjectMetadata
640 2 : var err error
641 2 : objMeta, err = provider.Lookup(fileTypeTable, f.FileBacking.DiskFileNum)
642 2 : if err != nil {
643 0 : return err
644 0 : }
645 2 : if !objMeta.IsShared() {
646 0 : return errors.Wrapf(ErrInvalidSkipSharedIteration, "file %s is not shared", objMeta.DiskFileNum)
647 0 : }
648 2 : if !base.Visible(f.LargestSeqNum, seqNum, base.InternalKeySeqNumMax) {
649 1 : return errors.Wrapf(ErrInvalidSkipSharedIteration, "file %s contains keys newer than snapshot", objMeta.DiskFileNum)
650 1 : }
651 2 : var sst *SharedSSTMeta
652 2 : var skip bool
653 2 : sst, skip, err = iter.db.truncateSharedFile(ctx, lower, upper, level, f, objMeta)
654 2 : if err != nil {
655 0 : return err
656 0 : }
657 2 : if skip {
658 2 : continue
659 : }
660 2 : if err = opts.visitSharedFile(sst); err != nil {
661 0 : return err
662 0 : }
663 : }
664 : }
665 : }
666 :
667 2 : for valid := iter.seekGE(lower); valid && iter.error() == nil; valid = iter.next() {
668 2 : key := iter.unsafeKey()
669 2 :
670 2 : if opts.rateLimitFunc != nil {
671 0 : if err := opts.rateLimitFunc(key, iter.lazyValue()); err != nil {
672 0 : return err
673 0 : }
674 : }
675 :
676 2 : switch key.Kind() {
677 2 : case InternalKeyKindRangeKeyDelete, InternalKeyKindRangeKeyUnset, InternalKeyKindRangeKeySet:
678 2 : if opts.visitRangeKey != nil {
679 2 : span := iter.unsafeSpan()
680 2 : // NB: The caller isn't interested in the sequence numbers of these
681 2 : // range keys. Rather, the caller wants them to be in trailer order
682 2 : // _after_ zeroing of sequence numbers. Copy span.Keys, sort it, and then
683 2 : // call visitRangeKey.
684 2 : keysCopy := make([]keyspan.Key, len(span.Keys))
685 2 : for i := range span.Keys {
686 2 : keysCopy[i] = span.Keys[i]
687 2 : keysCopy[i].Trailer = base.MakeTrailer(0, span.Keys[i].Kind())
688 2 : }
689 2 : keyspan.SortKeysByTrailer(&keysCopy)
690 2 : if err := opts.visitRangeKey(span.Start, span.End, keysCopy); err != nil {
691 0 : return err
692 0 : }
693 : }
694 2 : case InternalKeyKindRangeDelete:
695 2 : if opts.visitRangeDel != nil {
696 2 : rangeDel := iter.unsafeRangeDel()
697 2 : if err := opts.visitRangeDel(rangeDel.Start, rangeDel.End, rangeDel.LargestSeqNum()); err != nil {
698 0 : return err
699 0 : }
700 : }
701 2 : default:
702 2 : if opts.visitPointKey != nil {
703 2 : var info IteratorLevel
704 2 : if len(iter.mergingIter.heap.items) > 0 {
705 2 : mergingIterIdx := iter.mergingIter.heap.items[0].index
706 2 : info = iter.iterLevels[mergingIterIdx]
707 2 : } else {
708 0 : info = IteratorLevel{Kind: IteratorLevelUnknown}
709 0 : }
710 2 : val := iter.lazyValue()
711 2 : if err := opts.visitPointKey(key, val, info); err != nil {
712 0 : return err
713 0 : }
714 : }
715 : }
716 : }
717 :
718 2 : return nil
719 : }
720 :
721 : // constructPointIter constructs a merging iterator and sets i.iter to it.
722 : func (i *scanInternalIterator) constructPointIter(
723 : categoryAndQoS sstable.CategoryAndQoS, memtables flushableList, buf *iterAlloc,
724 2 : ) {
725 2 : // Merging levels and levels from iterAlloc.
726 2 : mlevels := buf.mlevels[:0]
727 2 : levels := buf.levels[:0]
728 2 :
729 2 : // We compute the number of levels needed ahead of time and reallocate a slice if
730 2 : // the array from the iterAlloc isn't large enough. Doing this allocation once
731 2 : // should improve the performance.
732 2 : numMergingLevels := len(memtables)
733 2 : numLevelIters := 0
734 2 :
735 2 : current := i.version
736 2 : if current == nil {
737 2 : current = i.readState.current
738 2 : }
739 2 : numMergingLevels += len(current.L0SublevelFiles)
740 2 : numLevelIters += len(current.L0SublevelFiles)
741 2 :
742 2 : for level := 1; level < len(current.Levels); level++ {
743 2 : if current.Levels[level].Empty() {
744 2 : continue
745 : }
746 2 : if i.opts.skipSharedLevels && level >= sharedLevelsStart {
747 2 : continue
748 : }
749 2 : numMergingLevels++
750 2 : numLevelIters++
751 : }
752 :
753 2 : if numMergingLevels > cap(mlevels) {
754 1 : mlevels = make([]mergingIterLevel, 0, numMergingLevels)
755 1 : }
756 2 : if numLevelIters > cap(levels) {
757 1 : levels = make([]levelIter, 0, numLevelIters)
758 1 : }
759 : // TODO(bilal): Push these into the iterAlloc buf.
760 2 : var rangeDelMiter keyspanimpl.MergingIter
761 2 : rangeDelIters := make([]keyspan.FragmentIterator, 0, numMergingLevels)
762 2 : rangeDelLevels := make([]keyspanimpl.LevelIter, 0, numLevelIters)
763 2 :
764 2 : i.iterLevels = make([]IteratorLevel, numMergingLevels)
765 2 : mlevelsIndex := 0
766 2 :
767 2 : // Next are the memtables.
768 2 : for j := len(memtables) - 1; j >= 0; j-- {
769 2 : mem := memtables[j]
770 2 : mlevels = append(mlevels, mergingIterLevel{
771 2 : iter: mem.newIter(&i.opts.IterOptions),
772 2 : })
773 2 : i.iterLevels[mlevelsIndex] = IteratorLevel{
774 2 : Kind: IteratorLevelFlushable,
775 2 : FlushableIndex: j,
776 2 : }
777 2 : mlevelsIndex++
778 2 : if rdi := mem.newRangeDelIter(&i.opts.IterOptions); rdi != nil {
779 2 : rangeDelIters = append(rangeDelIters, rdi)
780 2 : }
781 : }
782 :
783 : // Next are the file levels: L0 sub-levels followed by lower levels.
784 2 : levelsIndex := len(levels)
785 2 : mlevels = mlevels[:numMergingLevels]
786 2 : levels = levels[:numLevelIters]
787 2 : rangeDelLevels = rangeDelLevels[:numLevelIters]
788 2 : i.opts.IterOptions.snapshotForHideObsoletePoints = i.seqNum
789 2 : i.opts.IterOptions.CategoryAndQoS = categoryAndQoS
790 2 : addLevelIterForFiles := func(files manifest.LevelIterator, level manifest.Level) {
791 2 : li := &levels[levelsIndex]
792 2 : rli := &rangeDelLevels[levelsIndex]
793 2 :
794 2 : li.init(
795 2 : i.ctx, i.opts.IterOptions, i.comparer, i.newIters, files, level,
796 2 : internalIterOpts{})
797 2 : li.initBoundaryContext(&mlevels[mlevelsIndex].levelIterBoundaryContext)
798 2 : mlevels[mlevelsIndex].iter = li
799 2 : rli.Init(keyspan.SpanIterOptions{RangeKeyFilters: i.opts.RangeKeyFilters},
800 2 : i.comparer.Compare, tableNewRangeDelIter(i.ctx, i.newIters), files, level,
801 2 : manifest.KeyTypePoint)
802 2 : rangeDelIters = append(rangeDelIters, rli)
803 2 :
804 2 : levelsIndex++
805 2 : mlevelsIndex++
806 2 : }
807 :
808 2 : for j := len(current.L0SublevelFiles) - 1; j >= 0; j-- {
809 2 : i.iterLevels[mlevelsIndex] = IteratorLevel{
810 2 : Kind: IteratorLevelLSM,
811 2 : Level: 0,
812 2 : Sublevel: j,
813 2 : }
814 2 : addLevelIterForFiles(current.L0SublevelFiles[j].Iter(), manifest.L0Sublevel(j))
815 2 : }
816 : // Add level iterators for the non-empty non-L0 levels.
817 2 : for level := 1; level < numLevels; level++ {
818 2 : if current.Levels[level].Empty() {
819 2 : continue
820 : }
821 2 : if i.opts.skipSharedLevels && level >= sharedLevelsStart {
822 2 : continue
823 : }
824 2 : i.iterLevels[mlevelsIndex] = IteratorLevel{Kind: IteratorLevelLSM, Level: level}
825 2 : addLevelIterForFiles(current.Levels[level].Iter(), manifest.Level(level))
826 : }
827 :
828 2 : buf.merging.init(&i.opts.IterOptions, &InternalIteratorStats{}, i.comparer.Compare, i.comparer.Split, mlevels...)
829 2 : buf.merging.snapshot = i.seqNum
830 2 : rangeDelMiter.Init(i.comparer.Compare, keyspan.VisibleTransform(i.seqNum), new(keyspanimpl.MergingBuffers), rangeDelIters...)
831 2 :
832 2 : if i.opts.includeObsoleteKeys {
833 1 : iiter := &keyspan.InterleavingIter{}
834 1 : iiter.Init(i.comparer, &buf.merging, &rangeDelMiter,
835 1 : keyspan.InterleavingIterOpts{
836 1 : LowerBound: i.opts.LowerBound,
837 1 : UpperBound: i.opts.UpperBound,
838 1 : })
839 1 : i.pointKeyIter = iiter
840 2 : } else {
841 2 : pcIter := &pointCollapsingIterator{
842 2 : comparer: i.comparer,
843 2 : merge: i.merge,
844 2 : seqNum: i.seqNum,
845 2 : }
846 2 : pcIter.iter.Init(i.comparer, &buf.merging, &rangeDelMiter, keyspan.InterleavingIterOpts{
847 2 : LowerBound: i.opts.LowerBound,
848 2 : UpperBound: i.opts.UpperBound,
849 2 : })
850 2 : i.pointKeyIter = pcIter
851 2 : }
852 2 : i.iter = i.pointKeyIter
853 : }
854 :
855 : // constructRangeKeyIter constructs the range-key iterator stack, populating
856 : // i.rangeKey.rangeKeyIter with the resulting iterator. This is similar to
857 : // Iterator.constructRangeKeyIter, except it doesn't handle batches and ensures
858 : // iterConfig does *not* elide unsets/deletes.
859 2 : func (i *scanInternalIterator) constructRangeKeyIter() error {
860 2 : // We want the bounded iter from iterConfig, but not the collapsing of
861 2 : // RangeKeyUnsets and RangeKeyDels.
862 2 : i.rangeKey.rangeKeyIter = i.rangeKey.iterConfig.Init(
863 2 : i.comparer, i.seqNum, i.opts.LowerBound, i.opts.UpperBound,
864 2 : nil /* hasPrefix */, nil /* prefix */, true, /* internalKeys */
865 2 : &i.rangeKey.rangeKeyBuffers.internal)
866 2 :
867 2 : // Next are the flushables: memtables and large batches.
868 2 : if i.readState != nil {
869 2 : for j := len(i.readState.memtables) - 1; j >= 0; j-- {
870 2 : mem := i.readState.memtables[j]
871 2 : // We only need to read from memtables which contain sequence numbers older
872 2 : // than seqNum.
873 2 : if logSeqNum := mem.logSeqNum; logSeqNum >= i.seqNum {
874 2 : continue
875 : }
876 2 : if rki := mem.newRangeKeyIter(&i.opts.IterOptions); rki != nil {
877 2 : i.rangeKey.iterConfig.AddLevel(rki)
878 2 : }
879 : }
880 : }
881 :
882 2 : current := i.version
883 2 : if current == nil {
884 2 : current = i.readState.current
885 2 : }
886 : // Next are the file levels: L0 sub-levels followed by lower levels.
887 : //
888 : // Add file-specific iterators for L0 files containing range keys. This is less
889 : // efficient than using levelIters for sublevels of L0 files containing
890 : // range keys, but range keys are expected to be sparse anyway, reducing the
891 : // cost benefit of maintaining a separate L0Sublevels instance for range key
892 : // files and then using it here.
893 : //
894 : // NB: We iterate L0's files in reverse order. They're sorted by
895 : // LargestSeqNum ascending, and we need to add them to the merging iterator
896 : // in LargestSeqNum descending to preserve the merging iterator's invariants
897 : // around Key Trailer order.
898 2 : iter := current.RangeKeyLevels[0].Iter()
899 2 : for f := iter.Last(); f != nil; f = iter.Prev() {
900 2 : spanIter, err := i.newIterRangeKey(f, i.opts.SpanIterOptions())
901 2 : if err != nil {
902 0 : return err
903 0 : }
904 2 : i.rangeKey.iterConfig.AddLevel(spanIter)
905 : }
906 :
907 : // Add level iterators for the non-empty non-L0 levels.
908 2 : for level := 1; level < len(current.RangeKeyLevels); level++ {
909 2 : if current.RangeKeyLevels[level].Empty() {
910 2 : continue
911 : }
912 2 : if i.opts.skipSharedLevels && level >= sharedLevelsStart {
913 2 : continue
914 : }
915 2 : li := i.rangeKey.iterConfig.NewLevelIter()
916 2 : spanIterOpts := i.opts.SpanIterOptions()
917 2 : li.Init(spanIterOpts, i.comparer.Compare, i.newIterRangeKey, current.RangeKeyLevels[level].Iter(),
918 2 : manifest.Level(level), manifest.KeyTypeRange)
919 2 : i.rangeKey.iterConfig.AddLevel(li)
920 : }
921 2 : return nil
922 : }
923 :
924 : // seekGE seeks this iterator to the first key that's greater than or equal
925 : // to the specified user key.
926 2 : func (i *scanInternalIterator) seekGE(key []byte) bool {
927 2 : i.iterKey, i.iterValue = i.iter.SeekGE(key, base.SeekGEFlagsNone)
928 2 : return i.iterKey != nil
929 2 : }
930 :
931 : // unsafeKey returns the unsafe InternalKey at the current position. The value
932 : // is nil if the iterator is invalid or exhausted.
933 2 : func (i *scanInternalIterator) unsafeKey() *InternalKey {
934 2 : return i.iterKey
935 2 : }
936 :
937 : // lazyValue returns a value pointer to the value at the current iterator
938 : // position. Behaviour undefined if unsafeKey() returns a Range key or Rangedel
939 : // kind key.
940 2 : func (i *scanInternalIterator) lazyValue() LazyValue {
941 2 : return i.iterValue
942 2 : }
943 :
944 : // unsafeRangeDel returns a range key span. Behaviour undefined if UnsafeKey returns
945 : // a non-rangedel kind.
946 2 : func (i *scanInternalIterator) unsafeRangeDel() *keyspan.Span {
947 2 : type spanInternalIterator interface {
948 2 : Span() *keyspan.Span
949 2 : }
950 2 : return i.pointKeyIter.(spanInternalIterator).Span()
951 2 : }
952 :
953 : // unsafeSpan returns a range key span. Behaviour undefined if UnsafeKey returns
954 : // a non-rangekey type.
955 2 : func (i *scanInternalIterator) unsafeSpan() *keyspan.Span {
956 2 : return i.rangeKey.iiter.Span()
957 2 : }
958 :
959 : // next advances the iterator in the forward direction, and returns the
960 : // iterator's new validity state.
961 2 : func (i *scanInternalIterator) next() bool {
962 2 : i.iterKey, i.iterValue = i.iter.Next()
963 2 : return i.iterKey != nil
964 2 : }
965 :
966 : // error returns an error from the internal iterator, if there's any.
967 2 : func (i *scanInternalIterator) error() error {
968 2 : return i.iter.Error()
969 2 : }
970 :
971 : // close closes this iterator, and releases any pooled objects.
972 2 : func (i *scanInternalIterator) close() error {
973 2 : if err := i.iter.Close(); err != nil {
974 0 : return err
975 0 : }
976 2 : if i.readState != nil {
977 2 : i.readState.unref()
978 2 : }
979 2 : if i.version != nil {
980 1 : i.version.Unref()
981 1 : }
982 2 : if i.rangeKey != nil {
983 2 : i.rangeKey.PrepareForReuse()
984 2 : *i.rangeKey = iteratorRangeKeyState{
985 2 : rangeKeyBuffers: i.rangeKey.rangeKeyBuffers,
986 2 : }
987 2 : iterRangeKeyStateAllocPool.Put(i.rangeKey)
988 2 : i.rangeKey = nil
989 2 : }
990 2 : if alloc := i.alloc; alloc != nil {
991 2 : for j := range i.boundsBuf {
992 2 : if cap(i.boundsBuf[j]) >= maxKeyBufCacheSize {
993 0 : alloc.boundsBuf[j] = nil
994 2 : } else {
995 2 : alloc.boundsBuf[j] = i.boundsBuf[j]
996 2 : }
997 : }
998 2 : *alloc = iterAlloc{
999 2 : keyBuf: alloc.keyBuf[:0],
1000 2 : boundsBuf: alloc.boundsBuf,
1001 2 : prefixOrFullSeekKey: alloc.prefixOrFullSeekKey[:0],
1002 2 : }
1003 2 : iterAllocPool.Put(alloc)
1004 2 : i.alloc = nil
1005 : }
1006 2 : return nil
1007 : }
1008 :
1009 2 : func (i *scanInternalIterator) initializeBoundBufs(lower, upper []byte) {
1010 2 : buf := i.boundsBuf[i.boundsBufIdx][:0]
1011 2 : if lower != nil {
1012 2 : buf = append(buf, lower...)
1013 2 : i.opts.LowerBound = buf
1014 2 : } else {
1015 1 : i.opts.LowerBound = nil
1016 1 : }
1017 2 : if upper != nil {
1018 2 : buf = append(buf, upper...)
1019 2 : i.opts.UpperBound = buf[len(buf)-len(upper):]
1020 2 : } else {
1021 1 : i.opts.UpperBound = nil
1022 1 : }
1023 2 : i.boundsBuf[i.boundsBufIdx] = buf
1024 2 : i.boundsBufIdx = 1 - i.boundsBufIdx
1025 : }
|
