Line data Source code
1 : // Copyright 2019 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 : "io"
11 : "sort"
12 :
13 : "github.com/cockroachdb/errors"
14 : "github.com/cockroachdb/pebble/internal/base"
15 : "github.com/cockroachdb/pebble/internal/keyspan"
16 : "github.com/cockroachdb/pebble/internal/manifest"
17 : )
18 :
19 : // This file implements DB.CheckLevels() which checks that every entry in the
20 : // DB is consistent with respect to the level invariant: any point (or the
21 : // infinite number of points in a range tombstone) has a seqnum such that a
22 : // point with the same UserKey at a lower level has a lower seqnum. This is an
23 : // expensive check since it involves iterating over all the entries in the DB,
24 : // hence only intended for tests or tools.
25 : //
26 : // If we ignore range tombstones, the consistency checking of points can be
27 : // done with a simplified version of mergingIter. simpleMergingIter is that
28 : // simplified version of mergingIter that only needs to step through points
29 : // (analogous to only doing Next()). It can also easily accommodate
30 : // consistency checking of points relative to range tombstones.
31 : // simpleMergingIter does not do any seek optimizations present in mergingIter
32 : // (it minimally needs to seek the range delete iterators to position them at
33 : // or past the current point) since it does not want to miss points for
34 : // purposes of consistency checking.
35 : //
36 : // Mutual consistency of range tombstones is non-trivial to check. One needs
37 : // to detect inversions of the form [a, c)#8 at higher level and [b, c)#10 at
38 : // a lower level. The start key of the former is not contained in the latter
39 : // and we can't use the exclusive end key, c, for a containment check since it
40 : // is the sentinel key. We observe that if these tombstones were fragmented
41 : // wrt each other we would have [a, b)#8 and [b, c)#8 at the higher level and
42 : // [b, c)#10 at the lower level and then it is is trivial to compare the two
43 : // [b, c) tombstones. Note that this fragmentation needs to take into account
44 : // that tombstones in a file may be untruncated and need to act within the
45 : // bounds of the file. This checking is performed by checkRangeTombstones()
46 : // and its helper functions.
47 :
48 : // The per-level structure used by simpleMergingIter.
49 : type simpleMergingIterLevel struct {
50 : iter internalIterator
51 : rangeDelIter keyspan.FragmentIterator
52 :
53 : iterKV *base.InternalKV
54 : tombstone *keyspan.Span
55 : }
56 :
57 : type simpleMergingIter struct {
58 : levels []simpleMergingIterLevel
59 : snapshot uint64
60 : heap simpleMergingIterHeap
61 : // The last point's key and level. For validation.
62 : lastKey InternalKey
63 : lastLevel int
64 : lastIterMsg string
65 : // A non-nil valueMerger means MERGE record processing is ongoing.
66 : valueMerger base.ValueMerger
67 : // The first error will cause step() to return false.
68 : err error
69 : numPoints int64
70 : merge Merge
71 : formatKey base.FormatKey
72 : }
73 :
74 : func (m *simpleMergingIter) init(
75 : merge Merge,
76 : cmp Compare,
77 : snapshot uint64,
78 : formatKey base.FormatKey,
79 : levels ...simpleMergingIterLevel,
80 1 : ) {
81 1 : m.levels = levels
82 1 : m.formatKey = formatKey
83 1 : m.merge = merge
84 1 : m.snapshot = snapshot
85 1 : m.lastLevel = -1
86 1 : m.heap.cmp = cmp
87 1 : m.heap.items = make([]simpleMergingIterItem, 0, len(levels))
88 1 : for i := range m.levels {
89 1 : l := &m.levels[i]
90 1 : l.iterKV = l.iter.First()
91 1 : if l.iterKV != nil {
92 1 : item := simpleMergingIterItem{
93 1 : index: i,
94 1 : value: l.iterKV.V,
95 1 : }
96 1 : item.key = l.iterKV.K.Clone()
97 1 : m.heap.items = append(m.heap.items, item)
98 1 : }
99 : }
100 1 : m.heap.init()
101 1 :
102 1 : if m.heap.len() == 0 {
103 1 : return
104 1 : }
105 1 : m.positionRangeDels()
106 : }
107 :
108 : // Positions all the rangedel iterators at or past the current top of the
109 : // heap, using SeekGE().
110 1 : func (m *simpleMergingIter) positionRangeDels() {
111 1 : item := &m.heap.items[0]
112 1 : for i := range m.levels {
113 1 : l := &m.levels[i]
114 1 : if l.rangeDelIter == nil {
115 1 : continue
116 : }
117 1 : t, err := l.rangeDelIter.SeekGE(item.key.UserKey)
118 1 : m.err = firstError(m.err, err)
119 1 : l.tombstone = t
120 : }
121 : }
122 :
123 : // Returns true if not yet done.
124 1 : func (m *simpleMergingIter) step() bool {
125 1 : if m.heap.len() == 0 || m.err != nil {
126 1 : return false
127 1 : }
128 1 : item := &m.heap.items[0]
129 1 : l := &m.levels[item.index]
130 1 : // Sentinels are not relevant for this point checking.
131 1 : if !item.key.IsExclusiveSentinel() && item.key.Visible(m.snapshot, base.InternalKeySeqNumMax) {
132 1 : // This is a visible point key.
133 1 : if !m.handleVisiblePoint(item, l) {
134 0 : return false
135 0 : }
136 : }
137 :
138 : // The iterator for the current level may be closed in the following call to
139 : // Next(). We save its debug string for potential use after it is closed -
140 : // either in this current step() invocation or on the next invocation.
141 1 : m.lastIterMsg = l.iter.String()
142 1 :
143 1 : // Step to the next point.
144 1 : l.iterKV = l.iter.Next()
145 1 : if l.iterKV == nil {
146 1 : m.err = errors.CombineErrors(l.iter.Error(), l.iter.Close())
147 1 : l.iter = nil
148 1 : m.heap.pop()
149 1 : } else {
150 1 : // Check point keys in an sstable are ordered. Although not required, we check
151 1 : // for memtables as well. A subtle check here is that successive sstables of
152 1 : // L1 and higher levels are ordered. This happens when levelIter moves to the
153 1 : // next sstable in the level, in which case item.key is previous sstable's
154 1 : // last point key.
155 1 : if !l.iterKV.K.IsExclusiveSentinel() && base.InternalCompare(m.heap.cmp, item.key, l.iterKV.K) >= 0 {
156 0 : m.err = errors.Errorf("out of order keys %s >= %s in %s",
157 0 : item.key.Pretty(m.formatKey), l.iterKV.K.Pretty(m.formatKey), l.iter)
158 0 : return false
159 0 : }
160 1 : item.key = base.InternalKey{
161 1 : Trailer: l.iterKV.K.Trailer,
162 1 : UserKey: append(item.key.UserKey[:0], l.iterKV.K.UserKey...),
163 1 : }
164 1 : item.value = l.iterKV.V
165 1 : if m.heap.len() > 1 {
166 1 : m.heap.fix(0)
167 1 : }
168 : }
169 1 : if m.err != nil {
170 0 : return false
171 0 : }
172 1 : if m.heap.len() == 0 {
173 1 : // If m.valueMerger != nil, the last record was a MERGE record.
174 1 : if m.valueMerger != nil {
175 1 : var closer io.Closer
176 1 : var err error
177 1 : _, closer, err = m.valueMerger.Finish(true /* includesBase */)
178 1 : if closer != nil {
179 0 : err = errors.CombineErrors(err, closer.Close())
180 0 : }
181 1 : if err != nil {
182 0 : m.err = errors.CombineErrors(m.err,
183 0 : errors.Wrapf(err, "merge processing error on key %s in %s",
184 0 : item.key.Pretty(m.formatKey), m.lastIterMsg))
185 0 : }
186 1 : m.valueMerger = nil
187 : }
188 1 : return false
189 : }
190 1 : m.positionRangeDels()
191 1 : return true
192 : }
193 :
194 : // handleVisiblePoint returns true if validation succeeded and level checking
195 : // can continue.
196 : func (m *simpleMergingIter) handleVisiblePoint(
197 : item *simpleMergingIterItem, l *simpleMergingIterLevel,
198 1 : ) (ok bool) {
199 1 : m.numPoints++
200 1 : keyChanged := m.heap.cmp(item.key.UserKey, m.lastKey.UserKey) != 0
201 1 : if !keyChanged {
202 1 : // At the same user key. We will see them in decreasing seqnum
203 1 : // order so the lastLevel must not be lower.
204 1 : if m.lastLevel > item.index {
205 0 : m.err = errors.Errorf("found InternalKey %s in %s and InternalKey %s in %s",
206 0 : item.key.Pretty(m.formatKey), l.iter, m.lastKey.Pretty(m.formatKey),
207 0 : m.lastIterMsg)
208 0 : return false
209 0 : }
210 1 : m.lastLevel = item.index
211 1 : } else {
212 1 : // The user key has changed.
213 1 : m.lastKey.Trailer = item.key.Trailer
214 1 : m.lastKey.UserKey = append(m.lastKey.UserKey[:0], item.key.UserKey...)
215 1 : m.lastLevel = item.index
216 1 : }
217 : // Ongoing series of MERGE records ends with a MERGE record.
218 1 : if keyChanged && m.valueMerger != nil {
219 1 : var closer io.Closer
220 1 : _, closer, m.err = m.valueMerger.Finish(true /* includesBase */)
221 1 : if m.err == nil && closer != nil {
222 0 : m.err = closer.Close()
223 0 : }
224 1 : m.valueMerger = nil
225 : }
226 1 : itemValue, _, err := item.value.Value(nil)
227 1 : if err != nil {
228 0 : m.err = err
229 0 : return false
230 0 : }
231 1 : if m.valueMerger != nil {
232 1 : // Ongoing series of MERGE records.
233 1 : switch item.key.Kind() {
234 1 : case InternalKeyKindSingleDelete, InternalKeyKindDelete, InternalKeyKindDeleteSized:
235 1 : var closer io.Closer
236 1 : _, closer, m.err = m.valueMerger.Finish(true /* includesBase */)
237 1 : if m.err == nil && closer != nil {
238 0 : m.err = closer.Close()
239 0 : }
240 1 : m.valueMerger = nil
241 1 : case InternalKeyKindSet, InternalKeyKindSetWithDelete:
242 1 : m.err = m.valueMerger.MergeOlder(itemValue)
243 1 : if m.err == nil {
244 1 : var closer io.Closer
245 1 : _, closer, m.err = m.valueMerger.Finish(true /* includesBase */)
246 1 : if m.err == nil && closer != nil {
247 0 : m.err = closer.Close()
248 0 : }
249 : }
250 1 : m.valueMerger = nil
251 1 : case InternalKeyKindMerge:
252 1 : m.err = m.valueMerger.MergeOlder(itemValue)
253 0 : default:
254 0 : m.err = errors.Errorf("pebble: invalid internal key kind %s in %s",
255 0 : item.key.Pretty(m.formatKey),
256 0 : l.iter)
257 0 : return false
258 : }
259 1 : } else if item.key.Kind() == InternalKeyKindMerge && m.err == nil {
260 1 : // New series of MERGE records.
261 1 : m.valueMerger, m.err = m.merge(item.key.UserKey, itemValue)
262 1 : }
263 1 : if m.err != nil {
264 0 : m.err = errors.Wrapf(m.err, "merge processing error on key %s in %s",
265 0 : item.key.Pretty(m.formatKey), l.iter)
266 0 : return false
267 0 : }
268 : // Is this point covered by a tombstone at a lower level? Note that all these
269 : // iterators must be positioned at a key > item.key.
270 1 : for level := item.index + 1; level < len(m.levels); level++ {
271 1 : lvl := &m.levels[level]
272 1 : if lvl.rangeDelIter == nil || lvl.tombstone.Empty() {
273 1 : continue
274 : }
275 1 : if lvl.tombstone.Contains(m.heap.cmp, item.key.UserKey) && lvl.tombstone.CoversAt(m.snapshot, item.key.SeqNum()) {
276 0 : m.err = errors.Errorf("tombstone %s in %s deletes key %s in %s",
277 0 : lvl.tombstone.Pretty(m.formatKey), lvl.iter, item.key.Pretty(m.formatKey),
278 0 : l.iter)
279 0 : return false
280 0 : }
281 : }
282 1 : return true
283 : }
284 :
285 : // Checking that range tombstones are mutually consistent is performed by
286 : // checkRangeTombstones(). See the overview comment at the top of the file.
287 : //
288 : // We do this check as follows:
289 : // - Collect the tombstones for each level, put them into one pool of tombstones
290 : // along with their level information (addTombstonesFromIter()).
291 : // - Collect the start and end user keys from all these tombstones
292 : // (collectAllUserKey()) and use them to fragment all the tombstones
293 : // (fragmentUsingUserKey()).
294 : // - Sort tombstones by start key and decreasing seqnum
295 : // (tombstonesByStartKeyAndSeqnum) - all tombstones that have the same start
296 : // key will have the same end key because they have been fragmented.
297 : // - Iterate and check (iterateAndCheckTombstones()).
298 : //
299 : // Note that this simple approach requires holding all the tombstones across all
300 : // levels in-memory. A more sophisticated incremental approach could be devised,
301 : // if necessary.
302 :
303 : // A tombstone and the corresponding level it was found in.
304 : type tombstoneWithLevel struct {
305 : keyspan.Span
306 : level int
307 : // The level in LSM. A -1 means it's a memtable.
308 : lsmLevel int
309 : fileNum FileNum
310 : }
311 :
312 : // For sorting tombstoneWithLevels in increasing order of start UserKey and
313 : // for the same start UserKey in decreasing order of seqnum.
314 : type tombstonesByStartKeyAndSeqnum struct {
315 : cmp Compare
316 : buf []tombstoneWithLevel
317 : }
318 :
319 1 : func (v *tombstonesByStartKeyAndSeqnum) Len() int { return len(v.buf) }
320 1 : func (v *tombstonesByStartKeyAndSeqnum) Less(i, j int) bool {
321 1 : less := v.cmp(v.buf[i].Start, v.buf[j].Start)
322 1 : if less == 0 {
323 1 : return v.buf[i].LargestSeqNum() > v.buf[j].LargestSeqNum()
324 1 : }
325 1 : return less < 0
326 : }
327 1 : func (v *tombstonesByStartKeyAndSeqnum) Swap(i, j int) {
328 1 : v.buf[i], v.buf[j] = v.buf[j], v.buf[i]
329 1 : }
330 :
331 : func iterateAndCheckTombstones(
332 : cmp Compare, formatKey base.FormatKey, tombstones []tombstoneWithLevel,
333 1 : ) error {
334 1 : sortBuf := tombstonesByStartKeyAndSeqnum{
335 1 : cmp: cmp,
336 1 : buf: tombstones,
337 1 : }
338 1 : sort.Sort(&sortBuf)
339 1 :
340 1 : // For a sequence of tombstones that share the same start UserKey, we will
341 1 : // encounter them in non-increasing seqnum order and so should encounter them
342 1 : // in non-decreasing level order.
343 1 : lastTombstone := tombstoneWithLevel{}
344 1 : for _, t := range tombstones {
345 1 : if cmp(lastTombstone.Start, t.Start) == 0 && lastTombstone.level > t.level {
346 0 : return errors.Errorf("encountered tombstone %s in %s"+
347 0 : " that has a lower seqnum than the same tombstone in %s",
348 0 : t.Span.Pretty(formatKey), levelOrMemtable(t.lsmLevel, t.fileNum),
349 0 : levelOrMemtable(lastTombstone.lsmLevel, lastTombstone.fileNum))
350 0 : }
351 1 : lastTombstone = t
352 : }
353 1 : return nil
354 : }
355 :
356 : type checkConfig struct {
357 : logger Logger
358 : comparer *Comparer
359 : readState *readState
360 : newIters tableNewIters
361 : seqNum uint64
362 : stats *CheckLevelsStats
363 : merge Merge
364 : formatKey base.FormatKey
365 : }
366 :
367 : // cmp is shorthand for comparer.Compare.
368 1 : func (c *checkConfig) cmp(a, b []byte) int { return c.comparer.Compare(a, b) }
369 :
370 1 : func checkRangeTombstones(c *checkConfig) error {
371 1 : var level int
372 1 : var tombstones []tombstoneWithLevel
373 1 : var err error
374 1 :
375 1 : memtables := c.readState.memtables
376 1 : for i := len(memtables) - 1; i >= 0; i-- {
377 1 : iter := memtables[i].newRangeDelIter(nil)
378 1 : if iter == nil {
379 1 : continue
380 : }
381 1 : tombstones, err = addTombstonesFromIter(
382 1 : iter, level, -1, 0, tombstones, c.seqNum, c.cmp, c.formatKey,
383 1 : )
384 1 : if err != nil {
385 0 : return err
386 0 : }
387 1 : level++
388 : }
389 :
390 1 : current := c.readState.current
391 1 : addTombstonesFromLevel := func(files manifest.LevelIterator, lsmLevel int) error {
392 1 : for f := files.First(); f != nil; f = files.Next() {
393 1 : lf := files.Take()
394 1 : iters, err := c.newIters(
395 1 : context.Background(), lf.FileMetadata, &IterOptions{level: manifest.Level(lsmLevel)},
396 1 : internalIterOpts{}, iterRangeDeletions)
397 1 : if err != nil {
398 0 : return err
399 0 : }
400 1 : if tombstones, err = addTombstonesFromIter(iters.RangeDeletion(), level, lsmLevel, f.FileNum,
401 1 : tombstones, c.seqNum, c.cmp, c.formatKey); err != nil {
402 0 : iters.CloseAll()
403 0 : return err
404 0 : }
405 1 : iters.CloseAll()
406 : }
407 1 : return nil
408 : }
409 : // Now the levels with untruncated tombsones.
410 1 : for i := len(current.L0SublevelFiles) - 1; i >= 0; i-- {
411 1 : if current.L0SublevelFiles[i].Empty() {
412 0 : continue
413 : }
414 1 : err := addTombstonesFromLevel(current.L0SublevelFiles[i].Iter(), 0)
415 1 : if err != nil {
416 0 : return err
417 0 : }
418 1 : level++
419 : }
420 1 : for i := 1; i < len(current.Levels); i++ {
421 1 : if err := addTombstonesFromLevel(current.Levels[i].Iter(), i); err != nil {
422 0 : return err
423 0 : }
424 1 : level++
425 : }
426 1 : if c.stats != nil {
427 1 : c.stats.NumTombstones = len(tombstones)
428 1 : }
429 : // We now have truncated tombstones.
430 : // Fragment them all.
431 1 : userKeys := collectAllUserKeys(c.cmp, tombstones)
432 1 : tombstones = fragmentUsingUserKeys(c.cmp, tombstones, userKeys)
433 1 : return iterateAndCheckTombstones(c.cmp, c.formatKey, tombstones)
434 : }
435 :
436 0 : func levelOrMemtable(lsmLevel int, fileNum FileNum) string {
437 0 : if lsmLevel == -1 {
438 0 : return "memtable"
439 0 : }
440 0 : return fmt.Sprintf("L%d: fileNum=%s", lsmLevel, fileNum)
441 : }
442 :
443 : func addTombstonesFromIter(
444 : iter keyspan.FragmentIterator,
445 : level int,
446 : lsmLevel int,
447 : fileNum FileNum,
448 : tombstones []tombstoneWithLevel,
449 : seqNum uint64,
450 : cmp Compare,
451 : formatKey base.FormatKey,
452 1 : ) (_ []tombstoneWithLevel, err error) {
453 1 : defer func() {
454 1 : err = firstError(err, iter.Close())
455 1 : }()
456 :
457 1 : var prevTombstone keyspan.Span
458 1 : tomb, err := iter.First()
459 1 : for ; tomb != nil; tomb, err = iter.Next() {
460 1 : t := tomb.Visible(seqNum)
461 1 : if t.Empty() {
462 1 : continue
463 : }
464 1 : t = t.DeepClone()
465 1 : // This is mainly a test for rangeDelV2 formatted blocks which are expected to
466 1 : // be ordered and fragmented on disk. But we anyways check for memtables,
467 1 : // rangeDelV1 as well.
468 1 : if cmp(prevTombstone.End, t.Start) > 0 {
469 0 : return nil, errors.Errorf("unordered or unfragmented range delete tombstones %s, %s in %s",
470 0 : prevTombstone.Pretty(formatKey), t.Pretty(formatKey), levelOrMemtable(lsmLevel, fileNum))
471 0 : }
472 1 : prevTombstone = t
473 1 :
474 1 : if !t.Empty() {
475 1 : tombstones = append(tombstones, tombstoneWithLevel{
476 1 : Span: t,
477 1 : level: level,
478 1 : lsmLevel: lsmLevel,
479 1 : fileNum: fileNum,
480 1 : })
481 1 : }
482 : }
483 1 : if err != nil {
484 0 : return nil, err
485 0 : }
486 1 : return tombstones, nil
487 : }
488 :
489 : type userKeysSort struct {
490 : cmp Compare
491 : buf [][]byte
492 : }
493 :
494 1 : func (v *userKeysSort) Len() int { return len(v.buf) }
495 1 : func (v *userKeysSort) Less(i, j int) bool {
496 1 : return v.cmp(v.buf[i], v.buf[j]) < 0
497 1 : }
498 1 : func (v *userKeysSort) Swap(i, j int) {
499 1 : v.buf[i], v.buf[j] = v.buf[j], v.buf[i]
500 1 : }
501 1 : func collectAllUserKeys(cmp Compare, tombstones []tombstoneWithLevel) [][]byte {
502 1 : keys := make([][]byte, 0, len(tombstones)*2)
503 1 : for _, t := range tombstones {
504 1 : keys = append(keys, t.Start)
505 1 : keys = append(keys, t.End)
506 1 : }
507 1 : sorter := userKeysSort{
508 1 : cmp: cmp,
509 1 : buf: keys,
510 1 : }
511 1 : sort.Sort(&sorter)
512 1 : var last, curr int
513 1 : for last, curr = -1, 0; curr < len(keys); curr++ {
514 1 : if last < 0 || cmp(keys[last], keys[curr]) != 0 {
515 1 : last++
516 1 : keys[last] = keys[curr]
517 1 : }
518 : }
519 1 : keys = keys[:last+1]
520 1 : return keys
521 : }
522 :
523 : func fragmentUsingUserKeys(
524 : cmp Compare, tombstones []tombstoneWithLevel, userKeys [][]byte,
525 1 : ) []tombstoneWithLevel {
526 1 : var buf []tombstoneWithLevel
527 1 : for _, t := range tombstones {
528 1 : // Find the first position with tombstone start < user key
529 1 : i := sort.Search(len(userKeys), func(i int) bool {
530 1 : return cmp(t.Start, userKeys[i]) < 0
531 1 : })
532 1 : for ; i < len(userKeys); i++ {
533 1 : if cmp(userKeys[i], t.End) >= 0 {
534 1 : break
535 : }
536 1 : tPartial := t
537 1 : tPartial.End = userKeys[i]
538 1 : buf = append(buf, tPartial)
539 1 : t.Start = userKeys[i]
540 : }
541 1 : buf = append(buf, t)
542 : }
543 1 : return buf
544 : }
545 :
546 : // CheckLevelsStats provides basic stats on points and tombstones encountered.
547 : type CheckLevelsStats struct {
548 : NumPoints int64
549 : NumTombstones int
550 : }
551 :
552 : // CheckLevels checks:
553 : // - Every entry in the DB is consistent with the level invariant. See the
554 : // comment at the top of the file.
555 : // - Point keys in sstables are ordered.
556 : // - Range delete tombstones in sstables are ordered and fragmented.
557 : // - Successful processing of all MERGE records.
558 1 : func (d *DB) CheckLevels(stats *CheckLevelsStats) error {
559 1 : // Grab and reference the current readState.
560 1 : readState := d.loadReadState()
561 1 : defer readState.unref()
562 1 :
563 1 : // Determine the seqnum to read at after grabbing the read state (current and
564 1 : // memtables) above.
565 1 : seqNum := d.mu.versions.visibleSeqNum.Load()
566 1 :
567 1 : checkConfig := &checkConfig{
568 1 : logger: d.opts.Logger,
569 1 : comparer: d.opts.Comparer,
570 1 : readState: readState,
571 1 : newIters: d.newIters,
572 1 : seqNum: seqNum,
573 1 : stats: stats,
574 1 : merge: d.merge,
575 1 : formatKey: d.opts.Comparer.FormatKey,
576 1 : }
577 1 : return checkLevelsInternal(checkConfig)
578 1 : }
579 :
580 1 : func checkLevelsInternal(c *checkConfig) (err error) {
581 1 : // Phase 1: Use a simpleMergingIter to step through all the points and ensure
582 1 : // that points with the same user key at different levels are not inverted
583 1 : // wrt sequence numbers and the same holds for tombstones that cover points.
584 1 : // To do this, one needs to construct a simpleMergingIter which is similar to
585 1 : // how one constructs a mergingIter.
586 1 :
587 1 : // Add mem tables from newest to oldest.
588 1 : var mlevels []simpleMergingIterLevel
589 1 : defer func() {
590 1 : for i := range mlevels {
591 1 : l := &mlevels[i]
592 1 : if l.iter != nil {
593 1 : err = firstError(err, l.iter.Close())
594 1 : l.iter = nil
595 1 : }
596 1 : if l.rangeDelIter != nil {
597 1 : err = firstError(err, l.rangeDelIter.Close())
598 1 : l.rangeDelIter = nil
599 1 : }
600 : }
601 : }()
602 :
603 1 : memtables := c.readState.memtables
604 1 : for i := len(memtables) - 1; i >= 0; i-- {
605 1 : mem := memtables[i]
606 1 : mlevels = append(mlevels, simpleMergingIterLevel{
607 1 : iter: mem.newIter(nil),
608 1 : rangeDelIter: mem.newRangeDelIter(nil),
609 1 : })
610 1 : }
611 :
612 1 : current := c.readState.current
613 1 : // Determine the final size for mlevels so that there are no more
614 1 : // reallocations. levelIter will hold a pointer to elements in mlevels.
615 1 : start := len(mlevels)
616 1 : for sublevel := len(current.L0SublevelFiles) - 1; sublevel >= 0; sublevel-- {
617 1 : if current.L0SublevelFiles[sublevel].Empty() {
618 0 : continue
619 : }
620 1 : mlevels = append(mlevels, simpleMergingIterLevel{})
621 : }
622 1 : for level := 1; level < len(current.Levels); level++ {
623 1 : if current.Levels[level].Empty() {
624 1 : continue
625 : }
626 1 : mlevels = append(mlevels, simpleMergingIterLevel{})
627 : }
628 1 : mlevelAlloc := mlevels[start:]
629 1 : // Add L0 files by sublevel.
630 1 : for sublevel := len(current.L0SublevelFiles) - 1; sublevel >= 0; sublevel-- {
631 1 : if current.L0SublevelFiles[sublevel].Empty() {
632 0 : continue
633 : }
634 1 : manifestIter := current.L0SublevelFiles[sublevel].Iter()
635 1 : iterOpts := IterOptions{logger: c.logger}
636 1 : li := &levelIter{}
637 1 : li.init(context.Background(), iterOpts, c.comparer, c.newIters, manifestIter,
638 1 : manifest.L0Sublevel(sublevel), internalIterOpts{})
639 1 : li.initRangeDel(&mlevelAlloc[0].rangeDelIter)
640 1 : mlevelAlloc[0].iter = li
641 1 : mlevelAlloc = mlevelAlloc[1:]
642 : }
643 1 : for level := 1; level < len(current.Levels); level++ {
644 1 : if current.Levels[level].Empty() {
645 1 : continue
646 : }
647 :
648 1 : iterOpts := IterOptions{logger: c.logger}
649 1 : li := &levelIter{}
650 1 : li.init(context.Background(), iterOpts, c.comparer, c.newIters,
651 1 : current.Levels[level].Iter(), manifest.Level(level), internalIterOpts{})
652 1 : li.initRangeDel(&mlevelAlloc[0].rangeDelIter)
653 1 : mlevelAlloc[0].iter = li
654 1 : mlevelAlloc = mlevelAlloc[1:]
655 : }
656 :
657 1 : mergingIter := &simpleMergingIter{}
658 1 : mergingIter.init(c.merge, c.cmp, c.seqNum, c.formatKey, mlevels...)
659 1 : for cont := mergingIter.step(); cont; cont = mergingIter.step() {
660 1 : }
661 1 : if err := mergingIter.err; err != nil {
662 0 : return err
663 0 : }
664 1 : if c.stats != nil {
665 1 : c.stats.NumPoints = mergingIter.numPoints
666 1 : }
667 :
668 : // Phase 2: Check that the tombstones are mutually consistent.
669 1 : return checkRangeTombstones(c)
670 : }
671 :
672 : type simpleMergingIterItem struct {
673 : index int
674 : key InternalKey
675 : value base.LazyValue
676 : }
677 :
678 : type simpleMergingIterHeap struct {
679 : cmp Compare
680 : reverse bool
681 : items []simpleMergingIterItem
682 : }
683 :
684 1 : func (h *simpleMergingIterHeap) len() int {
685 1 : return len(h.items)
686 1 : }
687 :
688 1 : func (h *simpleMergingIterHeap) less(i, j int) bool {
689 1 : ikey, jkey := h.items[i].key, h.items[j].key
690 1 : if c := h.cmp(ikey.UserKey, jkey.UserKey); c != 0 {
691 1 : if h.reverse {
692 0 : return c > 0
693 0 : }
694 1 : return c < 0
695 : }
696 1 : if h.reverse {
697 0 : return ikey.Trailer < jkey.Trailer
698 0 : }
699 1 : return ikey.Trailer > jkey.Trailer
700 : }
701 :
702 1 : func (h *simpleMergingIterHeap) swap(i, j int) {
703 1 : h.items[i], h.items[j] = h.items[j], h.items[i]
704 1 : }
705 :
706 : // init, fix, up and down are copied from the go stdlib.
707 1 : func (h *simpleMergingIterHeap) init() {
708 1 : // heapify
709 1 : n := h.len()
710 1 : for i := n/2 - 1; i >= 0; i-- {
711 1 : h.down(i, n)
712 1 : }
713 : }
714 :
715 1 : func (h *simpleMergingIterHeap) fix(i int) {
716 1 : if !h.down(i, h.len()) {
717 1 : h.up(i)
718 1 : }
719 : }
720 :
721 1 : func (h *simpleMergingIterHeap) pop() *simpleMergingIterItem {
722 1 : n := h.len() - 1
723 1 : h.swap(0, n)
724 1 : h.down(0, n)
725 1 : item := &h.items[n]
726 1 : h.items = h.items[:n]
727 1 : return item
728 1 : }
729 :
730 1 : func (h *simpleMergingIterHeap) up(j int) {
731 1 : for {
732 1 : i := (j - 1) / 2 // parent
733 1 : if i == j || !h.less(j, i) {
734 1 : break
735 : }
736 0 : h.swap(i, j)
737 0 : j = i
738 : }
739 : }
740 :
741 1 : func (h *simpleMergingIterHeap) down(i0, n int) bool {
742 1 : i := i0
743 1 : for {
744 1 : j1 := 2*i + 1
745 1 : if j1 >= n || j1 < 0 { // j1 < 0 after int overflow
746 1 : break
747 : }
748 1 : j := j1 // left child
749 1 : if j2 := j1 + 1; j2 < n && h.less(j2, j1) {
750 1 : j = j2 // = 2*i + 2 // right child
751 1 : }
752 1 : if !h.less(j, i) {
753 1 : break
754 : }
755 1 : h.swap(i, j)
756 1 : i = j
757 : }
758 1 : return i > i0
759 : }
|
