Line data Source code
1 : // Copyright 2018 The LevelDB-Go and Pebble Authors. All rights reserved. Use
2 : // of this source code is governed by a BSD-style license that can be found in
3 : // the LICENSE file.
4 :
5 : package pebble
6 :
7 : import (
8 : "bytes"
9 : "context"
10 : "fmt"
11 : "runtime/debug"
12 : "unsafe"
13 :
14 : "github.com/cockroachdb/errors"
15 : "github.com/cockroachdb/pebble/internal/base"
16 : "github.com/cockroachdb/pebble/internal/invariants"
17 : "github.com/cockroachdb/pebble/internal/keyspan"
18 : )
19 :
20 : type mergingIterLevel struct {
21 : index int
22 : iter internalIterator
23 : // rangeDelIter is set to the range-deletion iterator for the level. When
24 : // configured with a levelIter, this pointer changes as sstable boundaries
25 : // are crossed. See levelIter.initRangeDel and the Range Deletions comment
26 : // below.
27 : rangeDelIter keyspan.FragmentIterator
28 : // rangeDelIterGeneration is incremented whenever rangeDelIter changes.
29 : rangeDelIterGeneration int
30 : // iterKV caches the current key-value pair iter points to.
31 : iterKV *base.InternalKV
32 : // levelIter is non-nil if this level's iter is ultimately backed by a
33 : // *levelIter. The handle in iter may have wrapped the levelIter with
34 : // intermediary internalIterator implementations.
35 : levelIter *levelIter
36 :
37 : // tombstone caches the tombstone rangeDelIter is currently pointed at. If
38 : // tombstone is nil, there are no further tombstones within the
39 : // current sstable in the current iterator direction. The cached tombstone is
40 : // only valid for the levels in the range [0,heap[0].index]. This avoids
41 : // positioning tombstones at lower levels which cannot possibly shadow the
42 : // current key.
43 : tombstone *keyspan.Span
44 : }
45 :
46 1 : func (ml *mergingIterLevel) setRangeDelIter(iter keyspan.FragmentIterator) {
47 1 : ml.tombstone = nil
48 1 : if ml.rangeDelIter != nil {
49 1 : ml.rangeDelIter.Close()
50 1 : }
51 1 : ml.rangeDelIter = iter
52 1 : ml.rangeDelIterGeneration++
53 : }
54 :
55 : // mergingIter provides a merged view of multiple iterators from different
56 : // levels of the LSM.
57 : //
58 : // The core of a mergingIter is a heap of internalIterators (see
59 : // mergingIterHeap). The heap can operate as either a min-heap, used during
60 : // forward iteration (First, SeekGE, Next) or a max-heap, used during reverse
61 : // iteration (Last, SeekLT, Prev). The heap is initialized in calls to First,
62 : // Last, SeekGE, and SeekLT. A call to Next or Prev takes the current top
63 : // element on the heap, advances its iterator, and then "fixes" the heap
64 : // property. When one of the child iterators is exhausted during Next/Prev
65 : // iteration, it is removed from the heap.
66 : //
67 : // # Range Deletions
68 : //
69 : // A mergingIter can optionally be configured with a slice of range deletion
70 : // iterators. The range deletion iterator slice must exactly parallel the point
71 : // iterators and the range deletion iterator must correspond to the same level
72 : // in the LSM as the point iterator. Note that each memtable and each table in
73 : // L0 is a different "level" from the mergingIter perspective. So level 0 below
74 : // does not correspond to L0 in the LSM.
75 : //
76 : // A range deletion iterator iterates over fragmented range tombstones. Range
77 : // tombstones are fragmented by splitting them at any overlapping points. This
78 : // fragmentation guarantees that within an sstable tombstones will either be
79 : // distinct or will have identical start and end user keys. While range
80 : // tombstones are fragmented within an sstable, the start and end keys are not truncated
81 : // to sstable boundaries. This is necessary because the tombstone end key is
82 : // exclusive and does not have a sequence number. Consider an sstable
83 : // containing the range tombstone [a,c)#9 and the key "b#8". The tombstone must
84 : // delete "b#8", yet older versions of "b" might spill over to the next
85 : // sstable. So the boundary key for this sstable must be "b#8". Adjusting the
86 : // end key of tombstones to be optionally inclusive or contain a sequence
87 : // number would be possible solutions (such solutions have potentially serious
88 : // issues: tombstones have exclusive end keys since an inclusive deletion end can
89 : // be converted to an exclusive one while the reverse transformation is not possible;
90 : // the semantics of a sequence number for the end key of a range tombstone are murky).
91 : //
92 : // The approach taken here performs an
93 : // implicit truncation of the tombstone to the sstable boundaries.
94 : //
95 : // During initialization of a mergingIter, the range deletion iterators for
96 : // batches, memtables, and L0 tables are populated up front. Note that Batches
97 : // and memtables index unfragmented tombstones. Batch.newRangeDelIter() and
98 : // memTable.newRangeDelIter() fragment and cache the tombstones on demand. The
99 : // L1-L6 range deletion iterators are populated by levelIter. When configured
100 : // to load range deletion iterators, whenever a levelIter loads a table it
101 : // loads both the point iterator and the range deletion
102 : // iterator. levelIter.rangeDelIter is configured to point to the right entry
103 : // in mergingIter.levels. The effect of this setup is that
104 : // mergingIter.levels[i].rangeDelIter always contains the fragmented range
105 : // tombstone for the current table in level i that the levelIter has open.
106 : //
107 : // Another crucial mechanism of levelIter is that it materializes fake point
108 : // entries for the table boundaries if the boundary is range deletion
109 : // key. Consider a table that contains only a range tombstone [a-e)#10. The
110 : // sstable boundaries for this table will be a#10,15 and
111 : // e#72057594037927935,15. During forward iteration levelIter will return
112 : // e#72057594037927935,15 as a key. During reverse iteration levelIter will
113 : // return a#10,15 as a key. These sentinel keys act as bookends to point
114 : // iteration and allow mergingIter to keep a table and its associated range
115 : // tombstones loaded as long as there are keys at lower levels that are within
116 : // the bounds of the table.
117 : //
118 : // The final piece to the range deletion puzzle is the LSM invariant that for a
119 : // given key K newer versions of K can only exist earlier in the level, or at
120 : // higher levels of the tree. For example, if K#4 exists in L3, k#5 can only
121 : // exist earlier in the L3 or in L0, L1, L2 or a memtable. Get very explicitly
122 : // uses this invariant to find the value for a key by walking the LSM level by
123 : // level. For range deletions, this invariant means that a range deletion at
124 : // level N will necessarily shadow any keys within its bounds in level Y where
125 : // Y > N. One wrinkle to this statement is that it only applies to keys that
126 : // lie within the sstable bounds as well, but we get that guarantee due to the
127 : // way the range deletion iterator and point iterator are bound together by a
128 : // levelIter.
129 : //
130 : // Tying the above all together, we get a picture where each level (index in
131 : // mergingIter.levels) is composed of both point operations (pX) and range
132 : // deletions (rX). The range deletions for level X shadow both the point
133 : // operations and range deletions for level Y where Y > X allowing mergingIter
134 : // to skip processing entries in that shadow. For example, consider the
135 : // scenario:
136 : //
137 : // r0: a---e
138 : // r1: d---h
139 : // r2: g---k
140 : // r3: j---n
141 : // r4: m---q
142 : //
143 : // This is showing 5 levels of range deletions. Consider what happens upon
144 : // SeekGE("b"). We first seek the point iterator for level 0 (the point values
145 : // are not shown above) and we then seek the range deletion iterator. That
146 : // returns the tombstone [a,e). This tombstone tells us that all keys in the
147 : // range [a,e) in lower levels are deleted so we can skip them. So we can
148 : // adjust the seek key to "e", the tombstone end key. For level 1 we seek to
149 : // "e" and find the range tombstone [d,h) and similar logic holds. By the time
150 : // we get to level 4 we're seeking to "n".
151 : //
152 : // One consequence of not truncating tombstone end keys to sstable boundaries
153 : // is the seeking process described above cannot always seek to the tombstone
154 : // end key in the older level. For example, imagine in the above example r3 is
155 : // a partitioned level (i.e., L1+ in our LSM), and the sstable containing [j,
156 : // n) has "k" as its upper boundary. In this situation, compactions involving
157 : // keys at or after "k" can output those keys to r4+, even if they're newer
158 : // than our tombstone [j, n). So instead of seeking to "n" in r4 we can only
159 : // seek to "k". To achieve this, the instance variable `largestUserKey.`
160 : // maintains the upper bounds of the current sstables in the partitioned
161 : // levels. In this example, `levels[3].largestUserKey` holds "k", telling us to
162 : // limit the seek triggered by a tombstone in r3 to "k".
163 : //
164 : // During actual iteration levels can contain both point operations and range
165 : // deletions. Within a level, when a range deletion contains a point operation
166 : // the sequence numbers must be checked to determine if the point operation is
167 : // newer or older than the range deletion tombstone. The mergingIter maintains
168 : // the invariant that the range deletion iterators for all levels newer that
169 : // the current iteration key (L < m.heap.items[0].index) are positioned at the
170 : // next (or previous during reverse iteration) range deletion tombstone. We
171 : // know those levels don't contain a range deletion tombstone that covers the
172 : // current key because if they did the current key would be deleted. The range
173 : // deletion iterator for the current key's level is positioned at a range
174 : // tombstone covering or past the current key. The position of all of other
175 : // range deletion iterators is unspecified. Whenever a key from those levels
176 : // becomes the current key, their range deletion iterators need to be
177 : // positioned. This lazy positioning avoids seeking the range deletion
178 : // iterators for keys that are never considered. (A similar bit of lazy
179 : // evaluation can be done for the point iterators, but is still TBD).
180 : //
181 : // For a full example, consider the following setup:
182 : //
183 : // p0: o
184 : // r0: m---q
185 : //
186 : // p1: n p
187 : // r1: g---k
188 : //
189 : // p2: b d i
190 : // r2: a---e q----v
191 : //
192 : // p3: e
193 : // r3:
194 : //
195 : // If we start iterating from the beginning, the first key we encounter is "b"
196 : // in p2. When the mergingIter is pointing at a valid entry, the range deletion
197 : // iterators for all of the levels < m.heap.items[0].index are positioned at
198 : // the next range tombstone past the current key. So r0 will point at [m,q) and
199 : // r1 at [g,k). When the key "b" is encountered, we check to see if the current
200 : // tombstone for r0 or r1 contains it, and whether the tombstone for r2, [a,e),
201 : // contains and is newer than "b".
202 : //
203 : // Advancing the iterator finds the next key at "d". This is in the same level
204 : // as the previous key "b" so we don't have to reposition any of the range
205 : // deletion iterators, but merely check whether "d" is now contained by any of
206 : // the range tombstones at higher levels or has stepped past the range
207 : // tombstone in its own level or higher levels. In this case, there is nothing to be done.
208 : //
209 : // Advancing the iterator again finds "e". Since "e" comes from p3, we have to
210 : // position the r3 range deletion iterator, which is empty. "e" is past the r2
211 : // tombstone of [a,e) so we need to advance the r2 range deletion iterator to
212 : // [q,v).
213 : //
214 : // The next key is "i". Because this key is in p2, a level above "e", we don't
215 : // have to reposition any range deletion iterators and instead see that "i" is
216 : // covered by the range tombstone [g,k). The iterator is immediately advanced
217 : // to "n" which is covered by the range tombstone [m,q) causing the iterator to
218 : // advance to "o" which is visible.
219 : //
220 : // # Error handling
221 : //
222 : // Any iterator operation may fail. The InternalIterator contract dictates that
223 : // an iterator must return a nil internal key when an error occurs, and a
224 : // subsequent call to Error() should return the error value. The exported
225 : // merging iterator positioning methods must adhere to this contract by setting
226 : // m.err to hold any error encountered by the individual level iterators and
227 : // returning a nil internal key. Some internal helpers (eg,
228 : // find[Next|Prev]Entry) also adhere to this contract, setting m.err directly).
229 : // Other internal functions return an explicit error return value and DO NOT set
230 : // m.err, relying on the caller to set m.err appropriately.
231 : //
232 : // TODO(jackson): Update the InternalIterator interface to return explicit error
233 : // return values (and an *InternalKV pointer).
234 : //
235 : // TODO(peter,rangedel): For testing, advance the iterator through various
236 : // scenarios and have each step display the current state (i.e. the current
237 : // heap and range-del iterator positioning).
238 : type mergingIter struct {
239 : logger Logger
240 : split Split
241 : dir int
242 : snapshot base.SeqNum
243 : batchSnapshot base.SeqNum
244 : levels []mergingIterLevel
245 : heap mergingIterHeap
246 : err error
247 : prefix []byte
248 : lower []byte
249 : upper []byte
250 : stats *InternalIteratorStats
251 : seekKeyBuf []byte
252 :
253 : // levelsPositioned, if non-nil, is a slice of the same length as levels.
254 : // It's used by NextPrefix to record which levels have already been
255 : // repositioned. It's created lazily by the first call to NextPrefix.
256 : levelsPositioned []bool
257 :
258 : combinedIterState *combinedIterState
259 :
260 : // Used in some tests to disable the random disabling of seek optimizations.
261 : forceEnableSeekOpt bool
262 : }
263 :
264 : // mergingIter implements the base.InternalIterator interface.
265 : var _ base.InternalIterator = (*mergingIter)(nil)
266 :
267 : // newMergingIter returns an iterator that merges its input. Walking the
268 : // resultant iterator will return all key/value pairs of all input iterators
269 : // in strictly increasing key order, as defined by cmp. It is permissible to
270 : // pass a nil split parameter if the caller is never going to call
271 : // SeekPrefixGE.
272 : //
273 : // The input's key ranges may overlap, but there are assumed to be no duplicate
274 : // keys: if iters[i] contains a key k then iters[j] will not contain that key k.
275 : //
276 : // None of the iters may be nil.
277 : func newMergingIter(
278 : logger Logger,
279 : stats *base.InternalIteratorStats,
280 : cmp Compare,
281 : split Split,
282 : iters ...internalIterator,
283 1 : ) *mergingIter {
284 1 : m := &mergingIter{}
285 1 : levels := make([]mergingIterLevel, len(iters))
286 1 : for i := range levels {
287 1 : levels[i].iter = iters[i]
288 1 : }
289 1 : m.init(&IterOptions{logger: logger}, stats, cmp, split, levels...)
290 1 : return m
291 : }
292 :
293 : func (m *mergingIter) init(
294 : opts *IterOptions,
295 : stats *base.InternalIteratorStats,
296 : cmp Compare,
297 : split Split,
298 : levels ...mergingIterLevel,
299 1 : ) {
300 1 : m.err = nil // clear cached iteration error
301 1 : m.logger = opts.getLogger()
302 1 : if opts != nil {
303 1 : m.lower = opts.LowerBound
304 1 : m.upper = opts.UpperBound
305 1 : }
306 1 : m.snapshot = base.SeqNumMax
307 1 : m.batchSnapshot = base.SeqNumMax
308 1 : m.levels = levels
309 1 : m.heap.cmp = cmp
310 1 : m.split = split
311 1 : m.stats = stats
312 1 : if cap(m.heap.items) < len(levels) {
313 1 : m.heap.items = make([]*mergingIterLevel, 0, len(levels))
314 1 : } else {
315 1 : m.heap.items = m.heap.items[:0]
316 1 : }
317 1 : for l := range m.levels {
318 1 : m.levels[l].index = l
319 1 : }
320 : }
321 :
322 1 : func (m *mergingIter) initHeap() {
323 1 : m.heap.items = m.heap.items[:0]
324 1 : for i := range m.levels {
325 1 : if l := &m.levels[i]; l.iterKV != nil {
326 1 : m.heap.items = append(m.heap.items, l)
327 1 : }
328 : }
329 1 : m.heap.init()
330 : }
331 :
332 1 : func (m *mergingIter) initMinHeap() error {
333 1 : m.dir = 1
334 1 : m.heap.reverse = false
335 1 : m.initHeap()
336 1 : return m.initMinRangeDelIters(-1)
337 1 : }
338 :
339 : // The level of the previous top element was oldTopLevel. Note that all range delete
340 : // iterators < oldTopLevel are positioned past the key of the previous top element and
341 : // the range delete iterator == oldTopLevel is positioned at or past the key of the
342 : // previous top element. We need to position the range delete iterators from oldTopLevel + 1
343 : // to the level of the current top element.
344 1 : func (m *mergingIter) initMinRangeDelIters(oldTopLevel int) error {
345 1 : if m.heap.len() == 0 {
346 1 : return nil
347 1 : }
348 :
349 : // Position the range-del iterators at levels <= m.heap.items[0].index.
350 1 : item := m.heap.items[0]
351 1 : for level := oldTopLevel + 1; level <= item.index; level++ {
352 1 : l := &m.levels[level]
353 1 : if l.rangeDelIter == nil {
354 1 : continue
355 : }
356 1 : var err error
357 1 : l.tombstone, err = l.rangeDelIter.SeekGE(item.iterKV.K.UserKey)
358 1 : if err != nil {
359 0 : return err
360 0 : }
361 : }
362 1 : return nil
363 : }
364 :
365 1 : func (m *mergingIter) initMaxHeap() error {
366 1 : m.dir = -1
367 1 : m.heap.reverse = true
368 1 : m.initHeap()
369 1 : return m.initMaxRangeDelIters(-1)
370 1 : }
371 :
372 : // The level of the previous top element was oldTopLevel. Note that all range delete
373 : // iterators < oldTopLevel are positioned before the key of the previous top element and
374 : // the range delete iterator == oldTopLevel is positioned at or before the key of the
375 : // previous top element. We need to position the range delete iterators from oldTopLevel + 1
376 : // to the level of the current top element.
377 1 : func (m *mergingIter) initMaxRangeDelIters(oldTopLevel int) error {
378 1 : if m.heap.len() == 0 {
379 1 : return nil
380 1 : }
381 : // Position the range-del iterators at levels <= m.heap.items[0].index.
382 1 : item := m.heap.items[0]
383 1 : for level := oldTopLevel + 1; level <= item.index; level++ {
384 1 : l := &m.levels[level]
385 1 : if l.rangeDelIter == nil {
386 1 : continue
387 : }
388 1 : tomb, err := keyspan.SeekLE(m.heap.cmp, l.rangeDelIter, item.iterKV.K.UserKey)
389 1 : if err != nil {
390 0 : return err
391 0 : }
392 1 : l.tombstone = tomb
393 : }
394 1 : return nil
395 : }
396 :
397 1 : func (m *mergingIter) switchToMinHeap() error {
398 1 : if m.heap.len() == 0 {
399 1 : if m.lower != nil {
400 0 : m.SeekGE(m.lower, base.SeekGEFlagsNone)
401 1 : } else {
402 1 : m.First()
403 1 : }
404 1 : return m.err
405 : }
406 :
407 : // We're switching from using a max heap to a min heap. We need to advance
408 : // any iterator that is less than or equal to the current key. Consider the
409 : // scenario where we have 2 iterators being merged (user-key:seq-num):
410 : //
411 : // i1: *a:2 b:2
412 : // i2: a:1 b:1
413 : //
414 : // The current key is a:2 and i2 is pointed at a:1. When we switch to forward
415 : // iteration, we want to return a key that is greater than a:2.
416 :
417 1 : key := m.heap.items[0].iterKV.K
418 1 : cur := m.heap.items[0]
419 1 :
420 1 : for i := range m.levels {
421 1 : l := &m.levels[i]
422 1 : if l == cur {
423 1 : continue
424 : }
425 1 : for l.iterKV = l.iter.Next(); l.iterKV != nil; l.iterKV = l.iter.Next() {
426 1 : if base.InternalCompare(m.heap.cmp, key, l.iterKV.K) < 0 {
427 1 : // key < iter-key
428 1 : break
429 : }
430 : // key >= iter-key
431 : }
432 1 : if l.iterKV == nil {
433 1 : if err := l.iter.Error(); err != nil {
434 0 : return err
435 0 : }
436 : }
437 : }
438 :
439 : // Special handling for the current iterator because we were using its key
440 : // above.
441 1 : cur.iterKV = cur.iter.Next()
442 1 : if cur.iterKV == nil {
443 1 : if err := cur.iter.Error(); err != nil {
444 0 : return err
445 0 : }
446 : }
447 1 : return m.initMinHeap()
448 : }
449 :
450 1 : func (m *mergingIter) switchToMaxHeap() error {
451 1 : if m.heap.len() == 0 {
452 1 : if m.upper != nil {
453 0 : m.SeekLT(m.upper, base.SeekLTFlagsNone)
454 1 : } else {
455 1 : m.Last()
456 1 : }
457 1 : return m.err
458 : }
459 :
460 : // We're switching from using a min heap to a max heap. We need to backup any
461 : // iterator that is greater than or equal to the current key. Consider the
462 : // scenario where we have 2 iterators being merged (user-key:seq-num):
463 : //
464 : // i1: a:2 *b:2
465 : // i2: a:1 b:1
466 : //
467 : // The current key is b:2 and i2 is pointing at b:1. When we switch to
468 : // reverse iteration, we want to return a key that is less than b:2.
469 1 : key := m.heap.items[0].iterKV.K
470 1 : cur := m.heap.items[0]
471 1 :
472 1 : for i := range m.levels {
473 1 : l := &m.levels[i]
474 1 : if l == cur {
475 1 : continue
476 : }
477 :
478 1 : for l.iterKV = l.iter.Prev(); l.iterKV != nil; l.iterKV = l.iter.Prev() {
479 1 : if base.InternalCompare(m.heap.cmp, key, l.iterKV.K) > 0 {
480 1 : // key > iter-key
481 1 : break
482 : }
483 : // key <= iter-key
484 : }
485 1 : if l.iterKV == nil {
486 1 : if err := l.iter.Error(); err != nil {
487 0 : return err
488 0 : }
489 : }
490 : }
491 :
492 : // Special handling for the current iterator because we were using its key
493 : // above.
494 1 : cur.iterKV = cur.iter.Prev()
495 1 : if cur.iterKV == nil {
496 1 : if err := cur.iter.Error(); err != nil {
497 0 : return err
498 0 : }
499 : }
500 1 : return m.initMaxHeap()
501 : }
502 :
503 : // nextEntry unconditionally steps to the next entry. item is the current top
504 : // item in the heap.
505 1 : func (m *mergingIter) nextEntry(l *mergingIterLevel, succKey []byte) error {
506 1 : // INVARIANT: If in prefix iteration mode, item.iterKey must have a prefix equal
507 1 : // to m.prefix. This invariant is important for ensuring TrySeekUsingNext
508 1 : // optimizations behave correctly.
509 1 : //
510 1 : // During prefix iteration, the iterator does not have a full view of the
511 1 : // LSM. Some level iterators may omit keys that are known to fall outside
512 1 : // the seek prefix (eg, due to sstable bloom filter exclusion). It's
513 1 : // important that in such cases we don't position any iterators beyond
514 1 : // m.prefix, because doing so may interfere with future seeks.
515 1 : //
516 1 : // Let prefixes P1 < P2 < P3. Imagine a SeekPrefixGE to prefix P1, followed
517 1 : // by a SeekPrefixGE to prefix P2. Imagine there exist live keys at prefix
518 1 : // P2, but they're not visible to the SeekPrefixGE(P1) (because of
519 1 : // bloom-filter exclusion or a range tombstone that deletes prefix P1 but
520 1 : // not P2). If the SeekPrefixGE(P1) is allowed to move any level iterators
521 1 : // to P3, the SeekPrefixGE(P2, TrySeekUsingNext=true) may mistakenly think
522 1 : // the level contains no point keys or range tombstones within the prefix
523 1 : // P2. Care is taken to avoid ever advancing the iterator beyond the current
524 1 : // prefix. If nextEntry is ever invoked while we're already beyond the
525 1 : // current prefix, we're violating the invariant.
526 1 : if invariants.Enabled && m.prefix != nil {
527 1 : if p := m.split.Prefix(l.iterKV.K.UserKey); !bytes.Equal(m.prefix, p) {
528 0 : m.logger.Fatalf("mergingIter: prefix violation: nexting beyond prefix %q; existing heap root %q\n%s",
529 0 : m.prefix, l.iterKV, debug.Stack())
530 0 : }
531 : }
532 :
533 1 : oldTopLevel := l.index
534 1 : oldRangeDelIterGeneration := l.rangeDelIterGeneration
535 1 :
536 1 : if succKey == nil {
537 1 : l.iterKV = l.iter.Next()
538 1 : } else {
539 1 : l.iterKV = l.iter.NextPrefix(succKey)
540 1 : }
541 :
542 1 : if l.iterKV == nil {
543 1 : if err := l.iter.Error(); err != nil {
544 0 : return err
545 0 : }
546 1 : m.heap.pop()
547 1 : } else {
548 1 : if m.prefix != nil && !bytes.Equal(m.prefix, m.split.Prefix(l.iterKV.K.UserKey)) {
549 1 : // Set keys without a matching prefix to their zero values when in prefix
550 1 : // iteration mode and remove iterated level from heap.
551 1 : l.iterKV = nil
552 1 : m.heap.pop()
553 1 : } else if m.heap.len() > 1 {
554 1 : m.heap.fix(0)
555 1 : }
556 1 : if l.rangeDelIterGeneration != oldRangeDelIterGeneration {
557 1 : // The rangeDelIter changed which indicates that the l.iter moved to the
558 1 : // next sstable. We have to update the tombstone for oldTopLevel as well.
559 1 : oldTopLevel--
560 1 : }
561 : }
562 :
563 : // The cached tombstones are only valid for the levels
564 : // [0,oldTopLevel]. Updated the cached tombstones for any levels in the range
565 : // [oldTopLevel+1,heap[0].index].
566 1 : return m.initMinRangeDelIters(oldTopLevel)
567 : }
568 :
569 : // isNextEntryDeleted starts from the current entry (as the next entry) and if
570 : // it is deleted, moves the iterators forward as needed and returns true, else
571 : // it returns false. item is the top item in the heap. If any of the required
572 : // iterator operations error, the error is returned without updating m.err.
573 : //
574 : // During prefix iteration mode, isNextEntryDeleted will exhaust the iterator by
575 : // clearing the heap if the deleted key(s) extend beyond the iteration prefix
576 : // during prefix-iteration mode.
577 1 : func (m *mergingIter) isNextEntryDeleted(item *mergingIterLevel) (bool, error) {
578 1 : // Look for a range deletion tombstone containing item.iterKV at higher
579 1 : // levels (level < item.index). If we find such a range tombstone we know
580 1 : // it deletes the key in the current level. Also look for a range
581 1 : // deletion at the current level (level == item.index). If we find such a
582 1 : // range deletion we need to check whether it is newer than the current
583 1 : // entry.
584 1 : for level := 0; level <= item.index; level++ {
585 1 : l := &m.levels[level]
586 1 : if l.rangeDelIter == nil || l.tombstone == nil {
587 1 : // If l.tombstone is nil, there are no further tombstones
588 1 : // in the current sstable in the current (forward) iteration
589 1 : // direction.
590 1 : continue
591 : }
592 1 : if m.heap.cmp(l.tombstone.End, item.iterKV.K.UserKey) <= 0 {
593 1 : // The current key is at or past the tombstone end key.
594 1 : //
595 1 : // NB: for the case that this l.rangeDelIter is provided by a levelIter we know that
596 1 : // the levelIter must be positioned at a key >= item.iterKV. So it is sufficient to seek the
597 1 : // current l.rangeDelIter (since any range del iterators that will be provided by the
598 1 : // levelIter in the future cannot contain item.iterKV). Also, it is possible that we
599 1 : // will encounter parts of the range delete that should be ignored -- we handle that
600 1 : // below.
601 1 : var err error
602 1 : l.tombstone, err = l.rangeDelIter.SeekGE(item.iterKV.K.UserKey)
603 1 : if err != nil {
604 0 : return false, err
605 0 : }
606 : }
607 1 : if l.tombstone == nil {
608 1 : continue
609 : }
610 :
611 1 : if l.tombstone.VisibleAt(m.snapshot) && m.heap.cmp(l.tombstone.Start, item.iterKV.K.UserKey) <= 0 {
612 1 : if level < item.index {
613 1 : // We could also do m.seekGE(..., level + 1). The levels from
614 1 : // [level + 1, item.index) are already after item.iterKV so seeking them may be
615 1 : // wasteful.
616 1 :
617 1 : // We can seek up to tombstone.End.
618 1 : //
619 1 : // Progress argument: Since this file is at a higher level than item.iterKV we know
620 1 : // that the iterator in this file must be positioned within its bounds and at a key
621 1 : // X > item.iterKV (otherwise it would be the min of the heap). It is not
622 1 : // possible for X.UserKey == item.iterKV.UserKey, since it is incompatible with
623 1 : // X > item.iterKV (a lower version cannot be in a higher sstable), so it must be that
624 1 : // X.UserKey > item.iterKV.UserKey. Which means l.largestUserKey > item.key.UserKey.
625 1 : // We also know that l.tombstone.End > item.iterKV.UserKey. So the min of these,
626 1 : // seekKey, computed below, is > item.iterKV.UserKey, so the call to seekGE() will
627 1 : // make forward progress.
628 1 : m.seekKeyBuf = append(m.seekKeyBuf[:0], l.tombstone.End...)
629 1 : seekKey := m.seekKeyBuf
630 1 : // This seek is not directly due to a SeekGE call, so we don't know
631 1 : // enough about the underlying iterator positions, and so we keep the
632 1 : // try-seek-using-next optimization disabled. Additionally, if we're in
633 1 : // prefix-seek mode and a re-seek would have moved us past the original
634 1 : // prefix, we can remove all merging iter levels below the rangedel
635 1 : // tombstone's level and return immediately instead of re-seeking. This
636 1 : // is correct since those levels cannot provide a key that matches the
637 1 : // prefix, and is also visible. Additionally, this is important to make
638 1 : // subsequent `TrySeekUsingNext` work correctly, as a re-seek on a
639 1 : // different prefix could have resulted in this iterator skipping visible
640 1 : // keys at prefixes in between m.prefix and seekKey, that are currently
641 1 : // not in the heap due to a bloom filter mismatch.
642 1 : //
643 1 : // Additionally, we set the relative-seek flag. This is
644 1 : // important when iterating with lazy combined iteration. If
645 1 : // there's a range key between this level's current file and the
646 1 : // file the seek will land on, we need to detect it in order to
647 1 : // trigger construction of the combined iterator.
648 1 : if m.prefix != nil {
649 1 : if !bytes.Equal(m.prefix, m.split.Prefix(seekKey)) {
650 1 : for i := item.index; i < len(m.levels); i++ {
651 1 : // Remove this level from the heap. Setting iterKV
652 1 : // to nil should be sufficient for initMinHeap to
653 1 : // not re-initialize the heap with them in it. Other
654 1 : // fields in mergingIterLevel can remain as-is; the
655 1 : // iter/rangeDelIter needs to stay intact for future
656 1 : // trySeekUsingNexts to work, the level iter
657 1 : // boundary context is owned by the levelIter which
658 1 : // is not being repositioned, and any tombstones in
659 1 : // these levels will be irrelevant for us anyway.
660 1 : m.levels[i].iterKV = nil
661 1 : }
662 : // TODO(bilal): Consider a more efficient way of removing levels from
663 : // the heap without reinitializing all of it. This would likely
664 : // necessitate tracking the heap positions of each mergingIterHeap
665 : // item in the mergingIterLevel, and then swapping that item in the
666 : // heap with the last-positioned heap item, and shrinking the heap by
667 : // one.
668 1 : if err := m.initMinHeap(); err != nil {
669 0 : return false, err
670 0 : }
671 1 : return true, nil
672 : }
673 : }
674 1 : if err := m.seekGE(seekKey, item.index, base.SeekGEFlagsNone.EnableRelativeSeek()); err != nil {
675 0 : return false, err
676 0 : }
677 1 : return true, nil
678 : }
679 1 : if l.tombstone.CoversAt(m.snapshot, item.iterKV.SeqNum()) {
680 1 : if err := m.nextEntry(item, nil /* succKey */); err != nil {
681 0 : return false, err
682 0 : }
683 1 : return true, nil
684 : }
685 : }
686 : }
687 1 : return false, nil
688 : }
689 :
690 : // Starting from the current entry, finds the first (next) entry that can be returned.
691 : //
692 : // If an error occurs, m.err is updated to hold the error and findNextentry
693 : // returns a nil internal key.
694 1 : func (m *mergingIter) findNextEntry() *base.InternalKV {
695 1 : for m.heap.len() > 0 && m.err == nil {
696 1 : item := m.heap.items[0]
697 1 :
698 1 : // The levelIter internal iterator will interleave exclusive sentinel
699 1 : // keys to keep files open until their range deletions are no longer
700 1 : // necessary. Sometimes these are interleaved with the user key of a
701 1 : // file's largest key, in which case they may simply be stepped over to
702 1 : // move to the next file in the forward direction. Other times they're
703 1 : // interleaved at the user key of the user-iteration boundary, if that
704 1 : // falls within the bounds of a file. In the latter case, there are no
705 1 : // more keys < m.upper, and we can stop iterating.
706 1 : //
707 1 : // We perform a key comparison to differentiate between these two cases.
708 1 : // This key comparison is considered okay because it only happens for
709 1 : // sentinel keys. It may be eliminated after #2863.
710 1 : if m.levels[item.index].iterKV.K.IsExclusiveSentinel() {
711 1 : if m.upper != nil && m.heap.cmp(m.levels[item.index].iterKV.K.UserKey, m.upper) >= 0 {
712 1 : break
713 : }
714 : // This key is the largest boundary of a file and can be skipped now
715 : // that the file's range deletions are no longer relevant.
716 1 : m.err = m.nextEntry(item, nil /* succKey */)
717 1 : if m.err != nil {
718 0 : return nil
719 0 : }
720 1 : continue
721 : }
722 :
723 1 : m.addItemStats(item)
724 1 :
725 1 : // Check if the heap root key is deleted by a range tombstone in a
726 1 : // higher level. If it is, isNextEntryDeleted will advance the iterator
727 1 : // to a later key (through seeking or nexting).
728 1 : isDeleted, err := m.isNextEntryDeleted(item)
729 1 : if err != nil {
730 0 : m.err = err
731 0 : return nil
732 1 : } else if isDeleted {
733 1 : m.stats.PointsCoveredByRangeTombstones++
734 1 : continue
735 : }
736 :
737 : // Check if the key is visible at the iterator sequence numbers.
738 1 : if !item.iterKV.Visible(m.snapshot, m.batchSnapshot) {
739 1 : m.err = m.nextEntry(item, nil /* succKey */)
740 1 : if m.err != nil {
741 0 : return nil
742 0 : }
743 1 : continue
744 : }
745 :
746 : // The heap root is visible and not deleted by any range tombstones.
747 : // Return it.
748 1 : return item.iterKV
749 : }
750 1 : return nil
751 : }
752 :
753 : // Steps to the prev entry. item is the current top item in the heap.
754 1 : func (m *mergingIter) prevEntry(l *mergingIterLevel) error {
755 1 : oldTopLevel := l.index
756 1 : oldRangeDelIterGeneration := l.rangeDelIterGeneration
757 1 : if l.iterKV = l.iter.Prev(); l.iterKV != nil {
758 1 : if m.heap.len() > 1 {
759 1 : m.heap.fix(0)
760 1 : }
761 1 : if l.rangeDelIterGeneration != oldRangeDelIterGeneration && l.rangeDelIter != nil {
762 1 : // The rangeDelIter changed which indicates that the l.iter moved to the
763 1 : // previous sstable. We have to update the tombstone for oldTopLevel as
764 1 : // well.
765 1 : oldTopLevel--
766 1 : }
767 1 : } else {
768 1 : if err := l.iter.Error(); err != nil {
769 0 : return err
770 0 : }
771 1 : m.heap.pop()
772 : }
773 :
774 : // The cached tombstones are only valid for the levels
775 : // [0,oldTopLevel]. Updated the cached tombstones for any levels in the range
776 : // [oldTopLevel+1,heap[0].index].
777 1 : return m.initMaxRangeDelIters(oldTopLevel)
778 : }
779 :
780 : // isPrevEntryDeleted() starts from the current entry (as the prev entry) and if it is deleted,
781 : // moves the iterators backward as needed and returns true, else it returns false. item is the top
782 : // item in the heap.
783 1 : func (m *mergingIter) isPrevEntryDeleted(item *mergingIterLevel) (bool, error) {
784 1 : // Look for a range deletion tombstone containing item.iterKV at higher
785 1 : // levels (level < item.index). If we find such a range tombstone we know
786 1 : // it deletes the key in the current level. Also look for a range
787 1 : // deletion at the current level (level == item.index). If we find such a
788 1 : // range deletion we need to check whether it is newer than the current
789 1 : // entry.
790 1 : for level := 0; level <= item.index; level++ {
791 1 : l := &m.levels[level]
792 1 : if l.rangeDelIter == nil || l.tombstone == nil {
793 1 : // If l.tombstone is nil, there are no further tombstones
794 1 : // in the current sstable in the current (reverse) iteration
795 1 : // direction.
796 1 : continue
797 : }
798 1 : if m.heap.cmp(item.iterKV.K.UserKey, l.tombstone.Start) < 0 {
799 1 : // The current key is before the tombstone start key.
800 1 : //
801 1 : // NB: for the case that this l.rangeDelIter is provided by a levelIter we know that
802 1 : // the levelIter must be positioned at a key < item.iterKV. So it is sufficient to seek the
803 1 : // current l.rangeDelIter (since any range del iterators that will be provided by the
804 1 : // levelIter in the future cannot contain item.iterKV). Also, it is it is possible that we
805 1 : // will encounter parts of the range delete that should be ignored -- we handle that
806 1 : // below.
807 1 :
808 1 : tomb, err := keyspan.SeekLE(m.heap.cmp, l.rangeDelIter, item.iterKV.K.UserKey)
809 1 : if err != nil {
810 0 : return false, err
811 0 : }
812 1 : l.tombstone = tomb
813 : }
814 1 : if l.tombstone == nil {
815 1 : continue
816 : }
817 1 : if l.tombstone.VisibleAt(m.snapshot) && m.heap.cmp(l.tombstone.End, item.iterKV.K.UserKey) > 0 {
818 1 : if level < item.index {
819 1 : // We could also do m.seekLT(..., level + 1). The levels from
820 1 : // [level + 1, item.index) are already before item.iterKV so seeking them may be
821 1 : // wasteful.
822 1 :
823 1 : // We can seek up to tombstone.Start.UserKey.
824 1 : //
825 1 : // Progress argument: We know that the iterator in this file is positioned within
826 1 : // its bounds and at a key X < item.iterKV (otherwise it would be the max of the heap).
827 1 : // So smallestUserKey <= item.iterKV.UserKey and we already know that
828 1 : // l.tombstone.Start.UserKey <= item.iterKV.UserKey. So the seekKey computed below
829 1 : // is <= item.iterKV.UserKey, and since we do a seekLT() we will make backwards
830 1 : // progress.
831 1 : m.seekKeyBuf = append(m.seekKeyBuf[:0], l.tombstone.Start...)
832 1 : seekKey := m.seekKeyBuf
833 1 : // We set the relative-seek flag. This is important when
834 1 : // iterating with lazy combined iteration. If there's a range
835 1 : // key between this level's current file and the file the seek
836 1 : // will land on, we need to detect it in order to trigger
837 1 : // construction of the combined iterator.
838 1 : if err := m.seekLT(seekKey, item.index, base.SeekLTFlagsNone.EnableRelativeSeek()); err != nil {
839 0 : return false, err
840 0 : }
841 1 : return true, nil
842 : }
843 1 : if l.tombstone.CoversAt(m.snapshot, item.iterKV.SeqNum()) {
844 1 : if err := m.prevEntry(item); err != nil {
845 0 : return false, err
846 0 : }
847 1 : return true, nil
848 : }
849 : }
850 : }
851 1 : return false, nil
852 : }
853 :
854 : // Starting from the current entry, finds the first (prev) entry that can be returned.
855 : //
856 : // If an error occurs, m.err is updated to hold the error and findNextentry
857 : // returns a nil internal key.
858 1 : func (m *mergingIter) findPrevEntry() *base.InternalKV {
859 1 : for m.heap.len() > 0 && m.err == nil {
860 1 : item := m.heap.items[0]
861 1 :
862 1 : // The levelIter internal iterator will interleave exclusive sentinel
863 1 : // keys to keep files open until their range deletions are no longer
864 1 : // necessary. Sometimes these are interleaved with the user key of a
865 1 : // file's smallest key, in which case they may simply be stepped over to
866 1 : // move to the next file in the backward direction. Other times they're
867 1 : // interleaved at the user key of the user-iteration boundary, if that
868 1 : // falls within the bounds of a file. In the latter case, there are no
869 1 : // more keys ≥ m.lower, and we can stop iterating.
870 1 : //
871 1 : // We perform a key comparison to differentiate between these two cases.
872 1 : // This key comparison is considered okay because it only happens for
873 1 : // sentinel keys. It may be eliminated after #2863.
874 1 : if m.levels[item.index].iterKV.K.IsExclusiveSentinel() {
875 1 : if m.lower != nil && m.heap.cmp(m.levels[item.index].iterKV.K.UserKey, m.lower) <= 0 {
876 1 : break
877 : }
878 : // This key is the smallest boundary of a file and can be skipped
879 : // now that the file's range deletions are no longer relevant.
880 1 : m.err = m.prevEntry(item)
881 1 : if m.err != nil {
882 0 : return nil
883 0 : }
884 1 : continue
885 : }
886 :
887 1 : m.addItemStats(item)
888 1 : if isDeleted, err := m.isPrevEntryDeleted(item); err != nil {
889 0 : m.err = err
890 0 : return nil
891 1 : } else if isDeleted {
892 1 : m.stats.PointsCoveredByRangeTombstones++
893 1 : continue
894 : }
895 1 : if item.iterKV.Visible(m.snapshot, m.batchSnapshot) {
896 1 : return item.iterKV
897 1 : }
898 1 : m.err = m.prevEntry(item)
899 : }
900 1 : return nil
901 : }
902 :
903 : // Seeks levels >= level to >= key. Additionally uses range tombstones to extend the seeks.
904 : //
905 : // If an error occurs, seekGE returns the error without setting m.err.
906 1 : func (m *mergingIter) seekGE(key []byte, level int, flags base.SeekGEFlags) error {
907 1 : // When seeking, we can use tombstones to adjust the key we seek to on each
908 1 : // level. Consider the series of range tombstones:
909 1 : //
910 1 : // 1: a---e
911 1 : // 2: d---h
912 1 : // 3: g---k
913 1 : // 4: j---n
914 1 : // 5: m---q
915 1 : //
916 1 : // If we SeekGE("b") we also find the tombstone "b" resides within in the
917 1 : // first level which is [a,e). Regardless of whether this tombstone deletes
918 1 : // "b" in that level, we know it deletes "b" in all lower levels, so we
919 1 : // adjust the search key in the next level to the tombstone end key "e". We
920 1 : // then SeekGE("e") in the second level and find the corresponding tombstone
921 1 : // [d,h). This process continues and we end up seeking for "h" in the 3rd
922 1 : // level, "k" in the 4th level and "n" in the last level.
923 1 : //
924 1 : // TODO(peter,rangedel): In addition to the above we can delay seeking a
925 1 : // level (and any lower levels) when the current iterator position is
926 1 : // contained within a range tombstone at a higher level.
927 1 :
928 1 : // Deterministically disable the TrySeekUsingNext optimizations sometimes in
929 1 : // invariant builds to encourage the metamorphic tests to surface bugs. Note
930 1 : // that we cannot disable the optimization within individual levels. It must
931 1 : // be disabled for all levels or none. If one lower-level iterator performs
932 1 : // a fresh seek whereas another takes advantage of its current iterator
933 1 : // position, the heap can become inconsistent. Consider the following
934 1 : // example:
935 1 : //
936 1 : // L5: [ [b-c) ] [ d ]*
937 1 : // L6: [ b ] [e]*
938 1 : //
939 1 : // Imagine a SeekGE(a). The [b-c) range tombstone deletes the L6 point key
940 1 : // 'b', resulting in the iterator positioned at d with the heap:
941 1 : //
942 1 : // {L5: d, L6: e}
943 1 : //
944 1 : // A subsequent SeekGE(b) is seeking to a larger key, so the caller may set
945 1 : // TrySeekUsingNext()=true. If the L5 iterator used the TrySeekUsingNext
946 1 : // optimization but the L6 iterator did not, the iterator would have the
947 1 : // heap:
948 1 : //
949 1 : // {L6: b, L5: d}
950 1 : //
951 1 : // Because the L5 iterator has already advanced to the next sstable, the
952 1 : // merging iterator cannot observe the [b-c) range tombstone and will
953 1 : // mistakenly return L6's deleted point key 'b'.
954 1 : if testingDisableSeekOpt(key, uintptr(unsafe.Pointer(m))) && !m.forceEnableSeekOpt {
955 1 : flags = flags.DisableTrySeekUsingNext()
956 1 : }
957 :
958 1 : for ; level < len(m.levels); level++ {
959 1 : if invariants.Enabled && m.lower != nil && m.heap.cmp(key, m.lower) < 0 {
960 0 : m.logger.Fatalf("mergingIter: lower bound violation: %s < %s\n%s", key, m.lower, debug.Stack())
961 0 : }
962 :
963 1 : l := &m.levels[level]
964 1 : if m.prefix != nil {
965 1 : l.iterKV = l.iter.SeekPrefixGE(m.prefix, key, flags)
966 1 : if l.iterKV != nil {
967 1 : if !bytes.Equal(m.prefix, m.split.Prefix(l.iterKV.K.UserKey)) {
968 1 : // Prevent keys without a matching prefix from being added to the heap by setting
969 1 : // iterKey and iterValue to their zero values before calling initMinHeap.
970 1 : l.iterKV = nil
971 1 : }
972 : }
973 1 : } else {
974 1 : l.iterKV = l.iter.SeekGE(key, flags)
975 1 : }
976 1 : if l.iterKV == nil {
977 1 : if err := l.iter.Error(); err != nil {
978 0 : return err
979 0 : }
980 : }
981 :
982 : // If this level contains overlapping range tombstones, alter the seek
983 : // key accordingly. Caveat: If we're performing lazy-combined iteration,
984 : // we cannot alter the seek key: Range tombstones don't delete range
985 : // keys, and there might exist live range keys within the range
986 : // tombstone's span that need to be observed to trigger a switch to
987 : // combined iteration.
988 1 : if rangeDelIter := l.rangeDelIter; rangeDelIter != nil &&
989 1 : (m.combinedIterState == nil || m.combinedIterState.initialized) {
990 1 : // The level has a range-del iterator. Find the tombstone containing
991 1 : // the search key.
992 1 : var err error
993 1 : l.tombstone, err = rangeDelIter.SeekGE(key)
994 1 : if err != nil {
995 0 : return err
996 0 : }
997 1 : if l.tombstone != nil && l.tombstone.VisibleAt(m.snapshot) && m.heap.cmp(l.tombstone.Start, key) <= 0 {
998 1 : // Based on the containment condition tombstone.End > key, so
999 1 : // the assignment to key results in a monotonically
1000 1 : // non-decreasing key across iterations of this loop.
1001 1 : //
1002 1 : // The adjustment of key here can only move it to a larger key.
1003 1 : // Since the caller of seekGE guaranteed that the original key
1004 1 : // was greater than or equal to m.lower, the new key will
1005 1 : // continue to be greater than or equal to m.lower.
1006 1 : key = l.tombstone.End
1007 1 : }
1008 : }
1009 : }
1010 1 : return m.initMinHeap()
1011 : }
1012 :
1013 0 : func (m *mergingIter) String() string {
1014 0 : return "merging"
1015 0 : }
1016 :
1017 : // SeekGE implements base.InternalIterator.SeekGE. Note that SeekGE only checks
1018 : // the upper bound. It is up to the caller to ensure that key is greater than
1019 : // or equal to the lower bound.
1020 1 : func (m *mergingIter) SeekGE(key []byte, flags base.SeekGEFlags) *base.InternalKV {
1021 1 : m.prefix = nil
1022 1 : m.err = m.seekGE(key, 0 /* start level */, flags)
1023 1 : if m.err != nil {
1024 0 : return nil
1025 0 : }
1026 1 : return m.findNextEntry()
1027 : }
1028 :
1029 : // SeekPrefixGE implements base.InternalIterator.SeekPrefixGE.
1030 1 : func (m *mergingIter) SeekPrefixGE(prefix, key []byte, flags base.SeekGEFlags) *base.InternalKV {
1031 1 : return m.SeekPrefixGEStrict(prefix, key, flags)
1032 1 : }
1033 :
1034 : // SeekPrefixGEStrict implements topLevelIterator.SeekPrefixGEStrict. Note that
1035 : // SeekPrefixGEStrict explicitly checks that the key has a matching prefix.
1036 : func (m *mergingIter) SeekPrefixGEStrict(
1037 : prefix, key []byte, flags base.SeekGEFlags,
1038 1 : ) *base.InternalKV {
1039 1 : m.prefix = prefix
1040 1 : m.err = m.seekGE(key, 0 /* start level */, flags)
1041 1 : if m.err != nil {
1042 0 : return nil
1043 0 : }
1044 :
1045 1 : iterKV := m.findNextEntry()
1046 1 : if invariants.Enabled && iterKV != nil {
1047 1 : if !bytes.Equal(m.prefix, m.split.Prefix(iterKV.K.UserKey)) {
1048 0 : m.logger.Fatalf("mergingIter: prefix violation: returning key %q without prefix %q\n", iterKV, m.prefix)
1049 0 : }
1050 : }
1051 1 : return iterKV
1052 : }
1053 :
1054 : // Seeks levels >= level to < key. Additionally uses range tombstones to extend the seeks.
1055 1 : func (m *mergingIter) seekLT(key []byte, level int, flags base.SeekLTFlags) error {
1056 1 : // See the comment in seekGE regarding using tombstones to adjust the seek
1057 1 : // target per level.
1058 1 : m.prefix = nil
1059 1 : for ; level < len(m.levels); level++ {
1060 1 : if invariants.Enabled && m.upper != nil && m.heap.cmp(key, m.upper) > 0 {
1061 0 : m.logger.Fatalf("mergingIter: upper bound violation: %s > %s\n%s", key, m.upper, debug.Stack())
1062 0 : }
1063 :
1064 1 : l := &m.levels[level]
1065 1 : l.iterKV = l.iter.SeekLT(key, flags)
1066 1 : if l.iterKV == nil {
1067 1 : if err := l.iter.Error(); err != nil {
1068 0 : return err
1069 0 : }
1070 : }
1071 :
1072 : // If this level contains overlapping range tombstones, alter the seek
1073 : // key accordingly. Caveat: If we're performing lazy-combined iteration,
1074 : // we cannot alter the seek key: Range tombstones don't delete range
1075 : // keys, and there might exist live range keys within the range
1076 : // tombstone's span that need to be observed to trigger a switch to
1077 : // combined iteration.
1078 1 : if rangeDelIter := l.rangeDelIter; rangeDelIter != nil &&
1079 1 : (m.combinedIterState == nil || m.combinedIterState.initialized) {
1080 1 : // The level has a range-del iterator. Find the tombstone containing
1081 1 : // the search key.
1082 1 : tomb, err := keyspan.SeekLE(m.heap.cmp, rangeDelIter, key)
1083 1 : if err != nil {
1084 0 : return err
1085 0 : }
1086 1 : l.tombstone = tomb
1087 1 : // Since SeekLT is exclusive on `key` and a tombstone's end key is
1088 1 : // also exclusive, a seek key equal to a tombstone's end key still
1089 1 : // enables the seek optimization (Note this is different than the
1090 1 : // check performed by (*keyspan.Span).Contains).
1091 1 : if l.tombstone != nil && l.tombstone.VisibleAt(m.snapshot) &&
1092 1 : m.heap.cmp(key, l.tombstone.End) <= 0 {
1093 1 : // NB: Based on the containment condition
1094 1 : // tombstone.Start.UserKey <= key, so the assignment to key
1095 1 : // results in a monotonically non-increasing key across
1096 1 : // iterations of this loop.
1097 1 : //
1098 1 : // The adjustment of key here can only move it to a smaller key.
1099 1 : // Since the caller of seekLT guaranteed that the original key
1100 1 : // was less than or equal to m.upper, the new key will continue
1101 1 : // to be less than or equal to m.upper.
1102 1 : key = l.tombstone.Start
1103 1 : }
1104 : }
1105 : }
1106 :
1107 1 : return m.initMaxHeap()
1108 : }
1109 :
1110 : // SeekLT implements base.InternalIterator.SeekLT. Note that SeekLT only checks
1111 : // the lower bound. It is up to the caller to ensure that key is less than the
1112 : // upper bound.
1113 1 : func (m *mergingIter) SeekLT(key []byte, flags base.SeekLTFlags) *base.InternalKV {
1114 1 : m.prefix = nil
1115 1 : m.err = m.seekLT(key, 0 /* start level */, flags)
1116 1 : if m.err != nil {
1117 0 : return nil
1118 0 : }
1119 1 : return m.findPrevEntry()
1120 : }
1121 :
1122 : // First implements base.InternalIterator.First. Note that First only checks
1123 : // the upper bound. It is up to the caller to ensure that key is greater than
1124 : // or equal to the lower bound (e.g. via a call to SeekGE(lower)).
1125 1 : func (m *mergingIter) First() *base.InternalKV {
1126 1 : m.err = nil // clear cached iteration error
1127 1 : m.prefix = nil
1128 1 : m.heap.items = m.heap.items[:0]
1129 1 : for i := range m.levels {
1130 1 : l := &m.levels[i]
1131 1 : l.iterKV = l.iter.First()
1132 1 : if l.iterKV == nil {
1133 1 : if m.err = l.iter.Error(); m.err != nil {
1134 0 : return nil
1135 0 : }
1136 : }
1137 : }
1138 1 : if m.err = m.initMinHeap(); m.err != nil {
1139 0 : return nil
1140 0 : }
1141 1 : return m.findNextEntry()
1142 : }
1143 :
1144 : // Last implements base.InternalIterator.Last. Note that Last only checks the
1145 : // lower bound. It is up to the caller to ensure that key is less than the
1146 : // upper bound (e.g. via a call to SeekLT(upper))
1147 1 : func (m *mergingIter) Last() *base.InternalKV {
1148 1 : m.err = nil // clear cached iteration error
1149 1 : m.prefix = nil
1150 1 : for i := range m.levels {
1151 1 : l := &m.levels[i]
1152 1 : l.iterKV = l.iter.Last()
1153 1 : if l.iterKV == nil {
1154 1 : if m.err = l.iter.Error(); m.err != nil {
1155 0 : return nil
1156 0 : }
1157 : }
1158 : }
1159 1 : if m.err = m.initMaxHeap(); m.err != nil {
1160 0 : return nil
1161 0 : }
1162 1 : return m.findPrevEntry()
1163 : }
1164 :
1165 1 : func (m *mergingIter) Next() *base.InternalKV {
1166 1 : if m.err != nil {
1167 0 : return nil
1168 0 : }
1169 :
1170 1 : if m.dir != 1 {
1171 1 : if m.err = m.switchToMinHeap(); m.err != nil {
1172 0 : return nil
1173 0 : }
1174 1 : return m.findNextEntry()
1175 : }
1176 :
1177 1 : if m.heap.len() == 0 {
1178 0 : return nil
1179 0 : }
1180 :
1181 : // NB: It's okay to call nextEntry directly even during prefix iteration
1182 : // mode. During prefix iteration mode, we rely on the caller to not call
1183 : // Next if the iterator has already advanced beyond the iteration prefix.
1184 : // See the comment above the base.InternalIterator interface.
1185 1 : if m.err = m.nextEntry(m.heap.items[0], nil /* succKey */); m.err != nil {
1186 0 : return nil
1187 0 : }
1188 :
1189 1 : iterKV := m.findNextEntry()
1190 1 : if invariants.Enabled && m.prefix != nil && iterKV != nil {
1191 1 : if !bytes.Equal(m.prefix, m.split.Prefix(iterKV.K.UserKey)) {
1192 0 : m.logger.Fatalf("mergingIter: prefix violation: returning key %q without prefix %q\n", iterKV, m.prefix)
1193 0 : }
1194 : }
1195 1 : return iterKV
1196 : }
1197 :
1198 1 : func (m *mergingIter) NextPrefix(succKey []byte) *base.InternalKV {
1199 1 : if m.dir != 1 {
1200 0 : panic("pebble: cannot switch directions with NextPrefix")
1201 : }
1202 1 : if m.err != nil || m.heap.len() == 0 {
1203 0 : return nil
1204 0 : }
1205 1 : if m.levelsPositioned == nil {
1206 1 : m.levelsPositioned = make([]bool, len(m.levels))
1207 1 : } else {
1208 1 : for i := range m.levelsPositioned {
1209 1 : m.levelsPositioned[i] = false
1210 1 : }
1211 : }
1212 :
1213 : // The heap root necessarily must be positioned at a key < succKey, because
1214 : // NextPrefix was invoked.
1215 1 : root := m.heap.items[0]
1216 1 : if invariants.Enabled && m.heap.cmp((*root).iterKV.K.UserKey, succKey) >= 0 {
1217 0 : m.logger.Fatalf("pebble: invariant violation: NextPrefix(%q) called on merging iterator already positioned at %q",
1218 0 : succKey, (*root).iterKV)
1219 0 : }
1220 : // NB: root is the heap root before we call nextEntry; nextEntry may change
1221 : // the heap root, so we must not `root` to still be the root of the heap, or
1222 : // even to be in the heap if the level's iterator becomes exhausted.
1223 1 : if m.err = m.nextEntry(root, succKey); m.err != nil {
1224 0 : return nil
1225 0 : }
1226 : // We only consider the level to be conclusively positioned at the next
1227 : // prefix if our call to nextEntry did not advance the level onto a range
1228 : // deletion's boundary. Range deletions may have bounds within the prefix
1229 : // that are still surfaced by NextPrefix.
1230 1 : m.levelsPositioned[root.index] = root.iterKV == nil || !root.iterKV.K.IsExclusiveSentinel()
1231 1 :
1232 1 : for m.heap.len() > 0 {
1233 1 : root := m.heap.items[0]
1234 1 : if m.levelsPositioned[root.index] {
1235 1 : // A level we've previously positioned is at the top of the heap, so
1236 1 : // there are no other levels positioned at keys < succKey. We've
1237 1 : // advanced as far as we need to.
1238 1 : break
1239 : }
1240 : // If the current heap root is a sentinel key, we need to skip it.
1241 : // Calling NextPrefix while positioned at a sentinel key is not
1242 : // supported.
1243 1 : if root.iterKV.K.IsExclusiveSentinel() {
1244 1 : if m.err = m.nextEntry(root, nil); m.err != nil {
1245 0 : return nil
1246 0 : }
1247 1 : continue
1248 : }
1249 :
1250 : // Since this level was not the original heap root when NextPrefix was
1251 : // called, we don't know whether this level's current key has the
1252 : // previous prefix or a new one.
1253 1 : if m.heap.cmp(root.iterKV.K.UserKey, succKey) >= 0 {
1254 1 : break
1255 : }
1256 1 : if m.err = m.nextEntry(root, succKey); m.err != nil {
1257 0 : return nil
1258 0 : }
1259 : // We only consider the level to be conclusively positioned at the next
1260 : // prefix if our call to nextEntry did not land onto a range deletion's
1261 : // boundary. Range deletions may have bounds within the prefix that are
1262 : // still surfaced by NextPrefix.
1263 1 : m.levelsPositioned[root.index] = root.iterKV == nil || !root.iterKV.K.IsExclusiveSentinel()
1264 : }
1265 1 : return m.findNextEntry()
1266 : }
1267 :
1268 1 : func (m *mergingIter) Prev() *base.InternalKV {
1269 1 : if m.err != nil {
1270 0 : return nil
1271 0 : }
1272 :
1273 1 : if m.dir != -1 {
1274 1 : if m.prefix != nil {
1275 0 : m.err = errors.New("pebble: unsupported reverse prefix iteration")
1276 0 : return nil
1277 0 : }
1278 1 : if m.err = m.switchToMaxHeap(); m.err != nil {
1279 0 : return nil
1280 0 : }
1281 1 : return m.findPrevEntry()
1282 : }
1283 :
1284 1 : if m.heap.len() == 0 {
1285 0 : return nil
1286 0 : }
1287 1 : if m.err = m.prevEntry(m.heap.items[0]); m.err != nil {
1288 0 : return nil
1289 0 : }
1290 1 : return m.findPrevEntry()
1291 : }
1292 :
1293 1 : func (m *mergingIter) Error() error {
1294 1 : if m.heap.len() == 0 || m.err != nil {
1295 1 : return m.err
1296 1 : }
1297 1 : return m.levels[m.heap.items[0].index].iter.Error()
1298 : }
1299 :
1300 1 : func (m *mergingIter) Close() error {
1301 1 : for i := range m.levels {
1302 1 : iter := m.levels[i].iter
1303 1 : if err := iter.Close(); err != nil && m.err == nil {
1304 0 : m.err = err
1305 0 : }
1306 1 : m.levels[i].setRangeDelIter(nil)
1307 : }
1308 1 : m.levels = nil
1309 1 : m.heap.items = m.heap.items[:0]
1310 1 : return m.err
1311 : }
1312 :
1313 1 : func (m *mergingIter) SetBounds(lower, upper []byte) {
1314 1 : m.prefix = nil
1315 1 : m.lower = lower
1316 1 : m.upper = upper
1317 1 : for i := range m.levels {
1318 1 : m.levels[i].iter.SetBounds(lower, upper)
1319 1 : }
1320 1 : m.heap.clear()
1321 : }
1322 :
1323 0 : func (m *mergingIter) SetContext(ctx context.Context) {
1324 0 : for i := range m.levels {
1325 0 : m.levels[i].iter.SetContext(ctx)
1326 0 : }
1327 : }
1328 :
1329 0 : func (m *mergingIter) DebugString() string {
1330 0 : var buf bytes.Buffer
1331 0 : sep := ""
1332 0 : for m.heap.len() > 0 {
1333 0 : item := m.heap.pop()
1334 0 : fmt.Fprintf(&buf, "%s%s", sep, item.iterKV.K)
1335 0 : sep = " "
1336 0 : }
1337 0 : var err error
1338 0 : if m.dir == 1 {
1339 0 : err = m.initMinHeap()
1340 0 : } else {
1341 0 : err = m.initMaxHeap()
1342 0 : }
1343 0 : if err != nil {
1344 0 : fmt.Fprintf(&buf, "err=<%s>", err)
1345 0 : }
1346 0 : return buf.String()
1347 : }
1348 :
1349 1 : func (m *mergingIter) ForEachLevelIter(fn func(li *levelIter) bool) {
1350 1 : for _, ml := range m.levels {
1351 1 : if ml.levelIter != nil {
1352 1 : if done := fn(ml.levelIter); done {
1353 1 : break
1354 : }
1355 : }
1356 : }
1357 : }
1358 :
1359 1 : func (m *mergingIter) addItemStats(l *mergingIterLevel) {
1360 1 : m.stats.PointCount++
1361 1 : m.stats.KeyBytes += uint64(len(l.iterKV.K.UserKey))
1362 1 : m.stats.ValueBytes += uint64(len(l.iterKV.V.ValueOrHandle))
1363 1 : }
1364 :
1365 : var _ internalIterator = &mergingIter{}
|
